Fix to support reading batched compressed messages

These messages span offsets, so "next" may sometimes be zero. KafkaContext changed to pass the actual message offset.
miniway · Apr 16, 2013 · 05129a7 · 05129a7
1 parent e65cfbf
commit 05129a7
Show file tree

Hide file tree

Showing 2 changed files with 3 additions and 2 deletions.
diff --git a/src/main/java/kafka/consumer/KafkaContext.java b/src/main/java/kafka/consumer/KafkaContext.java
@@ -125,9 +125,10 @@ public long getNext(LongWritable key, BytesWritable value) throws IOException {
 
         MessageAndOffset messageOffset = iterator.next();
         Message message = messageOffset.message();
+
+        key.set(curOffset);
         curOffset = messageOffset.offset();
 
-        key.set(curOffset - message.size() - 4);
         //byte[] bytes = new byte[message.payloadSize()];
         //message.payload().get(bytes);
         //value.set(bytes, 0, message.payloadSize());

diff --git a/src/main/java/kafka/consumer/KafkaInputFormat.java b/src/main/java/kafka/consumer/KafkaInputFormat.java
@@ -217,7 +217,7 @@ public boolean nextKeyValue() throws IOException, InterruptedException {
             if (limit < 0 || count < limit) {
 
                 long next = kcontext.getNext(key, value);
-                if (next > 0) {
+                if (next >= 0) {
                     pos = next;
                     count++;
                     return true;