You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ju...@apache.org on 2012/11/17 02:33:59 UTC
svn commit: r1410662 -
/incubator/kafka/branches/0.8/core/src/main/scala/kafka/tools/KafkaMigrationTool.java
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 **/