Index: core/src/main/scala/kafka/tools/KafkaMigrationTool.java
===================================================================
--- core/src/main/scala/kafka/tools/KafkaMigrationTool.java	(revision 1410624)
+++ core/src/main/scala/kafka/tools/KafkaMigrationTool.java	(working copy)
@@ -181,8 +181,9 @@
 
       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 **/
+      kafkaConsumerProperties_07.setProperty("shallowiterator.enable", "false");
       Object consumerConfig_07 = ConsumerConfigConstructor_07.newInstance(kafkaConsumerProperties_07);
 
       /** Construct the 07 consumer connector **/