Merge pull request apache#1073 from apache/upgradeToKafka22

SAMZA-2240:move samza-test to kafka 2.11-0.11

Author: sborya

samza-test/src/main/python/configs/downloads.json

@@ -1,5 +1,5 @@
 {
-  "url_kafka": "http://archive.apache.org/dist/kafka/0.10.1.1/kafka_2.10-0.10.1.1.tgz",
+  "url_kafka": "http://archive.apache.org/dist/kafka/0.11.0.3/kafka_2.11-0.11.0.3.tgz",
   "url_zookeeper": "http://archive.apache.org/dist/zookeeper/zookeeper-3.4.6/zookeeper-3.4.6.tar.gz",
   "url_hadoop": "https://archive.apache.org/dist/hadoop/common/hadoop-2.6.1/hadoop-2.6.1.tar.gz"
 }

samza-test/src/main/python/configs/kafka.json

@@ -3,21 +3,21 @@
     "kafka_instance_0": "localhost"
   },
   "kafka_port": 9092,
-  "kafka_start_cmd": "kafka_2.10-0.10.1.1/bin/kafka-server-start.sh -daemon kafka_2.10-0.10.1.1/config/server.properties --override delete.topic.enable=true",
-  "kafka_stop_cmd": "kafka_2.10-0.10.1.1/bin/kafka-server-stop.sh",
+  "kafka_start_cmd": "kafka_2.11-0.11.0.3/bin/kafka-server-start.sh -daemon kafka_2.11-0.11.0.3/config/server.properties --override delete.topic.enable=true",
+  "kafka_stop_cmd": "kafka_2.11-0.11.0.3/bin/kafka-server-stop.sh",
   "kafka_install_path": "deploy/kafka",
-  "kafka_executable": "kafka_2.10-0.10.1.1.tgz",
+  "kafka_executable": "kafka_2.11-0.11.0.3.tgz",
   "kafka_post_install_cmds": [
-    "sed -i.bak 's/SIGINT/SIGTERM/g' kafka_2.10-0.10.1.1/bin/kafka-server-stop.sh",
-    "sed -i.bak 's/^num\\.partitions *=.*/num.partitions=1/' kafka_2.10-0.10.1.1/config/server.properties",
-    "sed -i.bak 's/.*log.dirs.*/log.dirs=data/g' kafka_2.10-0.10.1.1/config/server.properties"
+    "sed -i.bak 's/SIGINT/SIGTERM/g' kafka_2.11-0.11.0.3/bin/kafka-server-stop.sh",
+    "sed -i.bak 's/^num\\.partitions *=.*/num.partitions=1/' kafka_2.11-0.11.0.3/config/server.properties",
+    "sed -i.bak 's/.*log.dirs.*/log.dirs=data/g' kafka_2.11-0.11.0.3/config/server.properties"
   ],
   "kafka_logs": [
     "log-cleaner.log",
-    "kafka_2.10-0.10.1.1/logs/controller.log",
-    "kafka_2.10-0.10.1.1/logs/kafka-request.log",
-    "kafka_2.10-0.10.1.1/logs/kafkaServer-gc.log",
-    "kafka_2.10-0.10.1.1/logs/server.log",
-    "kafka_2.10-0.10.1.1/logs/state-change.log"
+    "kafka_2.11-0.11.0.3/logs/controller.log",
+    "kafka_2.11-0.11.0.3/logs/kafka-request.log",
+    "kafka_2.11-0.11.0.3/logs/kafkaServer-gc.log",
+    "kafka_2.11-0.11.0.3/logs/server.log",
+    "kafka_2.11-0.11.0.3/logs/state-change.log"
   ]
 }

samza-test/src/main/python/standalone_deployment.py

@@ -93,7 +93,7 @@ def _create_kafka_topic(zookeeper_servers, topic_name, partition_count, replicat
 
     ### Using command line utility to create kafka topic since kafka python API doesn't support configuring partitionCount during topic creation.
     base_dir = os.getcwd()
-    create_topic_command = 'sh {0}/deploy/kafka/kafka_2.10-0.10.1.1/bin/kafka-topics.sh --create --zookeeper {1} --replication-factor {2} --partitions {3} --topic {4}'.format(base_dir, zookeeper_servers, replication_factor, partition_count, topic_name)
+    create_topic_command = 'sh {0}/deploy/kafka/kafka_2.11-0.11.0.3/bin/kafka-topics.sh --create --zookeeper {1} --replication-factor {2} --partitions {3} --topic {4}'.format(base_dir, zookeeper_servers, replication_factor, partition_count, topic_name)
     p = Popen(create_topic_command.split(' '), stdin=PIPE, stdout=PIPE, stderr=PIPE)
     output, err = p.communicate()
     logger.info("Output from create kafka topic: {0}\nstdout: {1}\nstderr: {2}".format(topic_name, output, err))
@@ -107,7 +107,7 @@ def _delete_kafka_topic(zookeeper_servers, topic_name):
     """
 
     base_dir = os.getcwd()
-    delete_topic_command = 'sh {0}/deploy/kafka/kafka_2.10-0.10.1.1/bin/kafka-topics.sh --delete --zookeeper {1} --topic {2}'.format(base_dir, zookeeper_servers, topic_name)
+    delete_topic_command = 'sh {0}/deploy/kafka/kafka_2.11-0.11.0.3/bin/kafka-topics.sh --delete --zookeeper {1} --topic {2}'.format(base_dir, zookeeper_servers, topic_name)
     logger.info("Deleting topic: {0}.".format(topic_name))
     p = Popen(delete_topic_command.split(' '), stdin=PIPE, stdout=PIPE, stderr=PIPE)
     output, err = p.communicate()