kafka-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jun...@apache.org
Subject svn commit: r1410662 - /incubator/kafka/branches/0.8/core/src/main/scala/kafka/tools/KafkaMigrationTool.java
Date Sat, 17 Nov 2012 01:33:59 GMT
Author: junrao
Date: Sat Nov 17 01:33:58 2012
New Revision: 1410662

URL: http://svn.apache.org/viewvc?rev=1410662&view=rev
Log:
MigrationTool should disable shallow iteration in the 0.7 consumer; patched by Yang Ye; reviewed
by Jun Rao; KAFKA-613

Modified:
    incubator/kafka/branches/0.8/core/src/main/scala/kafka/tools/KafkaMigrationTool.java

Modified: incubator/kafka/branches/0.8/core/src/main/scala/kafka/tools/KafkaMigrationTool.java
URL: http://svn.apache.org/viewvc/incubator/kafka/branches/0.8/core/src/main/scala/kafka/tools/KafkaMigrationTool.java?rev=1410662&r1=1410661&r2=1410662&view=diff
==============================================================================
--- incubator/kafka/branches/0.8/core/src/main/scala/kafka/tools/KafkaMigrationTool.java (original)
+++ incubator/kafka/branches/0.8/core/src/main/scala/kafka/tools/KafkaMigrationTool.java Sat
Nov 17 01:33:58 2012
@@ -19,7 +19,6 @@ package kafka.tools;
 
 import joptsimple.*;
 import kafka.javaapi.producer.Producer;
-import kafka.message.Message;
 import kafka.producer.KeyedMessage;
 import kafka.producer.ProducerConfig;
 import kafka.utils.Utils;
@@ -181,8 +180,12 @@ public class KafkaMigrationTool
 
       Constructor ConsumerConfigConstructor_07 = ConsumerConfig_07.getConstructor(Properties.class);
       Properties kafkaConsumerProperties_07 = new Properties();
-
       kafkaConsumerProperties_07.load(new FileInputStream(consumerConfigFile_07));
+      /** Disable shallow iteration because the message format is different between 07 and
08, we have to get each individual message **/
+      if(kafkaConsumerProperties_07.getProperty("shallowiterator.enable", "").equals("true")){
+        logger.warn("Shallow iterator should not be used in the migration tool");
+        kafkaConsumerProperties_07.setProperty("shallowiterator.enable", "false");
+      }
       Object consumerConfig_07 = ConsumerConfigConstructor_07.newInstance(kafkaConsumerProperties_07);
 
       /** Construct the 07 consumer connector **/



Mime
View raw message