[KAFKA-946] Kafka Hadoop Consumer fails when verifying message checksum

• Type: Bug
• Status: Closed
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 0.8.0
• Fix Version/s: 0.8.0
• Component/s: contrib
• Assignee: Sam Meder
• Reporter: Sam Meder
• Created: 18/Jun/13 18:28
• Updated: 24/Sep/13 15:32
• Resolved: 24/Sep/13 15:31

Description

The code tries to verify the checksum, but fails because the data available isn't the same. In KafkaETLContext:

protected boolean get(KafkaETLKey key, BytesWritable value) throws IOException {
if (_messageIt != null && _messageIt.hasNext())

{ MessageAndOffset messageAndOffset = _messageIt.next(); ByteBuffer buf = messageAndOffset.message().payload(); int origSize = buf.remaining(); byte[] bytes = new byte[origSize]; buf.get(bytes, buf.position(), origSize); value.set(bytes, 0, origSize); key.set(_index, _offset, messageAndOffset.message().checksum()); _offset = messageAndOffset.nextOffset(); //increase offset _count ++; //increase count return true; }

else return false;
}

Note that the message payload is used and the message checksum is included in the key. The in SimpleKafkaETLMapper:

@Override
public void map(KafkaETLKey key, BytesWritable val,
OutputCollector<LongWritable, Text> collector,
Reporter reporter) throws IOException {

byte[] bytes = KafkaETLUtils.getBytes(val);

//check the checksum of message
Message message = new Message(bytes);
long checksum = key.getChecksum();
if (checksum != message.checksum())
throw new IOException ("Invalid message checksum "
+ message.checksum() + ". Expected " + key + ".");

the Message object is initialized with the payload bytes and a new checksum is calculated. The problem is that the original message checksum also contains the key so checksum verification fails...