Merge remote-tracking branch 'upstream/master'

Conflicts:
	src/main/scripts/run_tests.sh

pom.xml

@@ -56,7 +56,7 @@
         <dependency>
             <groupId>org.apache.kafka</groupId>
             <artifactId>kafka_2.10</artifactId>
-            <version>0.8.1.1</version>
+            <version>0.8.2.1</version>
             <exclusions>
                 <exclusion>
                     <groupId>org.slf4j</groupId>

src/main/java/com/pinterest/secor/common/OffsetTracker.java

@@ -52,10 +52,10 @@ public long setLastSeenOffset(TopicPartition topicPartition, long offset) {
         mLastSeenOffset.put(topicPartition, offset);
         if (lastSeenOffset + 1 != offset) {
             if (lastSeenOffset >= 0) {
-                LOG.warn("offset for topic {} partition {}  changed from {} to {}",
+                LOG.warn("offset for topic {} partition {} changed from {} to {}",
                         topicPartition.getTopic(),topicPartition.getPartition(),lastSeenOffset, offset);
             } else {
-                LOG.info("starting to consume topic {} partition  from offset {}",
+                LOG.info("starting to consume topic {} partition {} from offset {}",
                         topicPartition.getTopic(),topicPartition.getPartition(),offset);
             }
         }

src/main/java/com/pinterest/secor/reader/MessageReader.java

@@ -110,6 +110,7 @@ private ConsumerConfig createConsumerConfig() throws UnknownHostException {
         props.put("auto.offset.reset", "smallest");
         props.put("consumer.timeout.ms", Integer.toString(mConfig.getConsumerTimeoutMs()));
         props.put("consumer.id", IdUtil.getConsumerId());
+        props.put("partition.assignment.strategy", "roundrobin");
         if (mConfig.getRebalanceMaxRetries() != null &&
             !mConfig.getRebalanceMaxRetries().isEmpty()) {
             props.put("rebalance.max.retries", mConfig.getRebalanceMaxRetries());

src/main/scripts/run_tests.sh

@@ -214,17 +214,14 @@ verify() {
 set_offsets_in_zookeeper() {
     for group in secor_backup secor_partition; do
         for partition in 0 1; do
-            run_command "${BASE_DIR}/run_zookeeper_command.sh localhost:2181 create \
-                /consumers \'\' > ${LOGS_DIR}/run_zookeeper_command.log 2>&1"
-            run_command "${BASE_DIR}/run_zookeeper_command.sh localhost:2181 create \
-                /consumers/${group} \'\' > ${LOGS_DIR}/run_zookeeper_command.log 2>&1"
-            run_command "${BASE_DIR}/run_zookeeper_command.sh localhost:2181 create \
-                /consumers/${group}/offsets \'\' > ${LOGS_DIR}/run_zookeeper_command.log 2>&1"
-            run_command "${BASE_DIR}/run_zookeeper_command.sh localhost:2181 create \
-                /consumers/${group}/offsets/test \'\' > ${LOGS_DIR}/run_zookeeper_command.log 2>&1"
-            run_command "${BASE_DIR}/run_zookeeper_command.sh localhost:2181 create \
-                /consumers/${group}/offsets/test/${partition} $1 > \
-                ${LOGS_DIR}/run_zookeeper_command.log 2>&1"
+            cat <<EOF | run_command "${BASE_DIR}/run_zookeeper_command.sh localhost:2181 > ${LOGS_DIR}/run_zookeeper_command.log 2>&1"
+create /consumers ''
+create /consumers/${group} ''
+create /consumers/${group}/offsets ''
+create /consumers/${group}/offsets/test ''
+create /consumers/${group}/offsets/test/${partition} $1
+quit
+EOF
         done
     done
 }

src/main/scripts/run_zookeeper_command.sh

@@ -17,11 +17,6 @@
 
 # Author: Pawel Garbacki ([email protected])
 
-if [ $# -lt 3 ]; then
-    echo "USAGE: $0 zookeeper_host:port cmd args"
-    exit 1
-fi
-
 CURR_DIR=`dirname $0`
 source ${CURR_DIR}/run_common.sh