Home | About | Sematext search-lucene.com search-hadoop.com
 Search Hadoop and all its subprojects:

Switch to Threaded View
Kafka, mail # user - Wire format for Kafka 0.7.1


Copy link to this message
-
Wire format for Kafka 0.7.1
Milind Parikh 2012-09-27, 06:21
I am writing an erlang driver for Kafka. I am using the spec from
https://cwiki.apache.org/confluence/display/KAFKA/Writing+a+Driver+for+Kafka.
Just learnt something that I thought should be the ML for someone
developing a different driver.

My specific issue currently has to do with the PRODUCE request. It appears
that the request header is getting parsed correctly. BUT the specific
message does not seem to parsed with the topic of "test" and the message of
"hi", partition 0, magic 0, compression 0.
[2012-09-26 22:55:30,131] INFO Created log for 'test'-0
(kafka.log.LogManager)
[2012-09-26 22:55:30,134] INFO Begin registering broker topic
/brokers/topics/test/0 with 1 partitions (kafka.server.KafkaZooKeeper)
[2012-09-26 22:55:30,138] ERROR Error processing ProduceRequest on test:0
(kafka.server.KafkaRequestHandlers)
kafka.message.InvalidMessageException: message is invalid, compression
codec: NoCompressionCodec size: 8 curr offset: 0 init offset: 0
    at
kafka.message.ByteBufferMessageSet$$anon$1.makeNextOuter(ByteBufferMessageSet.scala:130)
    at
kafka.message.ByteBufferMessageSet$$anon$1.makeNext(ByteBufferMessageSet.scala:166)
    at
kafka.message.ByteBufferMessageSet$$anon$1.makeNext(ByteBufferMessageSet.scala:100)
    at
kafka.utils.IteratorTemplate.maybeComputeNext(IteratorTemplate.scala:59)
    at kafka.utils.IteratorTemplate.hasNext(IteratorTemplate.scala:51)
    at scala.collection.Iterator$class.foreach(Iterator.scala:631)
    at kafka.utils.IteratorTemplate.foreach(IteratorTemplate.scala:30)
    at scala.collection.IterableLike$class.foreach(IterableLike.scala:79)
    at kafka.message.MessageSet.foreach(MessageSet.scala:87)
    at kafka.log.Log.append(Log.scala:205)
    at
kafka.server.KafkaRequestHandlers.kafka$server$KafkaRequestHandlers$$handleProducerRequest(KafkaRequestHandlers.scala:69)
    at
kafka.server.KafkaRequestHandlers.handleProducerRequest(KafkaRequestHandlers.scala:53)
    at
kafka.server.KafkaRequestHandlers$$anonfun$handlerFor$1.apply(KafkaRequestHandlers.scala:38)
    at
kafka.server.KafkaRequestHandlers$$anonfun$handlerFor$1.apply(KafkaRequestHandlers.scala:38)
    at kafka.network.Processor.handle(SocketServer.scala:296)
    at kafka.network.Processor.read(SocketServer.scala:319)
    at kafka.network.Processor.run(SocketServer.scala:214)
    at java.lang.Thread.run(Thread.java:679)
[2012-09-26 22:55:30,143] ERROR Closing socket for /127.0.0.1 because of
error (kafka.network.Processor)
kafka.message.InvalidMessageException: message is invalid, compression
codec: NoCompressionCodec size: 8 curr offset: 0 init offset: 0
    at
kafka.message.ByteBufferMessageSet$$anon$1.makeNextOuter(ByteBufferMessageSet.scala:130)
    at
kafka.message.ByteBufferMessageSet$$anon$1.makeNext(ByteBufferMessageSet.scala:166)
    at
kafka.message.ByteBufferMessageSet$$anon$1.makeNext(ByteBufferMessageSet.scala:100)
    at
kafka.utils.IteratorTemplate.maybeComputeNext(IteratorTemplate.scala:59)
    at kafka.utils.IteratorTemplate.hasNext(IteratorTemplate.scala:51)
    at scala.collection.Iterator$class.foreach(Iterator.scala:631)
    at kafka.utils.IteratorTemplate.foreach(IteratorTemplate.scala:30)
    at scala.collection.IterableLike$class.foreach(IterableLike.scala:79)
    at kafka.message.MessageSet.foreach(MessageSet.scala:87)
    at kafka.log.Log.append(Log.scala:205)
    at
kafka.server.KafkaRequestHandlers.kafka$server$KafkaRequestHandlers$$handleProducerRequest(KafkaRequestHandlers.scala:69)
    at
kafka.server.KafkaRequestHandlers.handleProducerRequest(KafkaRequestHandlers.scala:53)
    at
kafka.server.KafkaRequestHandlers$$anonfun$handlerFor$1.apply(KafkaRequestHandlers.scala:38)
    at
kafka.server.KafkaRequestHandlers$$anonfun$handlerFor$1.apply(KafkaRequestHandlers.scala:38)
    at kafka.network.Processor.handle(SocketServer.scala:296)
    at kafka.network.Processor.read(SocketServer.scala:319)
    at kafka.network.Processor.run(SocketServer.scala:214)
    at java.lang.Thread.run(Thread.java:679)
[2012-09-26 22:55:30,233] INFO End registering broker topic
/brokers/topics/test/0 (kafka.server.KafkaZooKeeper)
^C[2012-09-26 22:56:18,290] INFO Shutting down Kafka server
(kafka.server.KafkaServer)
[2012-09-26 22:56:18,292] INFO shutdown scheduler kafka-logcleaner-
(kafka.utils.KafkaScheduler)
[2012-09-26 22:56:18,304] INFO shutdown scheduler kafka-logflusher-
(kafka.utils.KafkaScheduler)
[2012-09-26 22:56:18,364] INFO Closing zookeeper client...
(kafka.server.KafkaZooKeeper)
[2012-09-26 22:56:18,364] INFO Terminate ZkClient event thread.
(org.I0Itec.zkclient.ZkEventThread)
[2012-09-26 22:56:18,364] INFO zkActor stopped (kafka.log.LogManager)
[2012-09-26 22:56:18,375] INFO EventThread shut down
(org.apache.zookeeper.ClientCnxn)
[2012-09-26 22:56:18,376] INFO Session: 0x13a0647aa7d0000 closed
(org.apache.zookeeper.ZooKeeper)
[2012-09-26 22:56:18,376] INFO Kafka server shut down completed
(kafka.server.KafkaServer)
BUT the specific message *does GET* parsed with the topic of "test" and the
message of "hi", partition 0, *magic 1*, compression 0. *It's a case of
RTFM. *
 I produce the following byte map.

|------- REQUEST HEADER ---------|-----MESSAGES_LENGTH ---|------MESSAGE
                 16 bytes                                       4
bytes                      12 bytes
REQUEST_HEADER : 16 bytes

             | -- REQUEST_LENGTH-|---RQ-TYPE  --|-TP_LENGTH -|-----TOPIC

             1          2          3           4          5         6
         7         8          9        10         11          12        13
      14     15    16

             0          0           0         28          0        0
   0         4         116      101      115         116        0         0
      0      0

t          e         s             t
MESSAGES_LENGTH : 4 bytes

             1          2          3           4

             0          0          0           12
MESSAGE :  12 BYTES

             |----- LENGTH --------------|           M        C
|-----------CRC32-----