kafka.message.InvalidMessageException

JIRA | Erik Onnen | 6 years ago
  1. 0

    Unfortunately I wasn't able to figure out exactly how I accomplished this, but somehow I managed to get one or more poison messages in my test topic using fairly simple single threaded SimpleProducer/SimpleConsumer code. In this state, I could see several hundreds of thousands of messages in the topic via JMX and adding new messages appeared to succeed from the API calls, however each add resulted in the following error: [2010-12-08 14:56:04,952] ERROR kafka.message.InvalidMessageException (kafka.network.Processor) kafka.message.InvalidMessageException at kafka.log.Log$$anonfun$append$1.apply(Log.scala:187) at kafka.log.Log$$anonfun$append$1.apply(Log.scala:185) at scala.collection.Iterator$class.foreach(Iterator.scala:631) at kafka.utils.IteratorTemplate.foreach(IteratorTemplate.scala:29) at scala.collection.IterableLike$class.foreach(IterableLike.scala:79) at kafka.message.MessageSet.foreach(MessageSet.scala:63) at kafka.log.Log.append(Log.scala:185) at kafka.server.KafkaRequestHandlers.handleProducerRequest(KafkaRequestHandlers.scala:56) at kafka.server.KafkaRequestHandlers$$anonfun$handlerFor$1.apply(KafkaRequestHandlers.scala:40) at kafka.server.KafkaRequestHandlers$$anonfun$handlerFor$1.apply(KafkaRequestHandlers.scala:40) at kafka.network.Processor.handle(SocketServer.scala:268) at kafka.network.Processor.read(SocketServer.scala:291) at kafka.network.Processor.run(SocketServer.scala:202) at java.lang.Thread.run(Thread.java:662) Additionally, no consumers were functional, they were all blocked and not draining the existing messages. Stopping the service did not get things healthy, but stopping and removing the logs fixed the problem. Not sure what the design intent is, but it would seem that failure to process a message internally within the server should cause the message to be rejected at the client layer.

    JIRA | 6 years ago | Erik Onnen
    kafka.message.InvalidMessageException
  2. 0

    Unfortunately I wasn't able to figure out exactly how I accomplished this, but somehow I managed to get one or more poison messages in my test topic using fairly simple single threaded SimpleProducer/SimpleConsumer code. In this state, I could see several hundreds of thousands of messages in the topic via JMX and adding new messages appeared to succeed from the API calls, however each add resulted in the following error: [2010-12-08 14:56:04,952] ERROR kafka.message.InvalidMessageException (kafka.network.Processor) kafka.message.InvalidMessageException at kafka.log.Log$$anonfun$append$1.apply(Log.scala:187) at kafka.log.Log$$anonfun$append$1.apply(Log.scala:185) at scala.collection.Iterator$class.foreach(Iterator.scala:631) at kafka.utils.IteratorTemplate.foreach(IteratorTemplate.scala:29) at scala.collection.IterableLike$class.foreach(IterableLike.scala:79) at kafka.message.MessageSet.foreach(MessageSet.scala:63) at kafka.log.Log.append(Log.scala:185) at kafka.server.KafkaRequestHandlers.handleProducerRequest(KafkaRequestHandlers.scala:56) at kafka.server.KafkaRequestHandlers$$anonfun$handlerFor$1.apply(KafkaRequestHandlers.scala:40) at kafka.server.KafkaRequestHandlers$$anonfun$handlerFor$1.apply(KafkaRequestHandlers.scala:40) at kafka.network.Processor.handle(SocketServer.scala:268) at kafka.network.Processor.read(SocketServer.scala:291) at kafka.network.Processor.run(SocketServer.scala:202) at java.lang.Thread.run(Thread.java:662) Additionally, no consumers were functional, they were all blocked and not draining the existing messages. Stopping the service did not get things healthy, but stopping and removing the logs fixed the problem. Not sure what the design intent is, but it would seem that failure to process a message internally within the server should cause the message to be rejected at the client layer.

    JIRA | 6 years ago | Erik Onnen
    kafka.message.InvalidMessageException

    Root Cause Analysis

    1. kafka.message.InvalidMessageException

      No message provided

      at kafka.log.Log$$anonfun$append$1.apply()
    2. Apache Kafka
      Log$$anonfun$append$1.apply
      1. kafka.log.Log$$anonfun$append$1.apply(Log.scala:187)
      2. kafka.log.Log$$anonfun$append$1.apply(Log.scala:185)
      2 frames
    3. Scala
      Iterator$class.foreach
      1. scala.collection.Iterator$class.foreach(Iterator.scala:631)
      1 frame
    4. Apache Kafka
      IteratorTemplate.foreach
      1. kafka.utils.IteratorTemplate.foreach(IteratorTemplate.scala:29)
      1 frame
    5. Scala
      IterableLike$class.foreach
      1. scala.collection.IterableLike$class.foreach(IterableLike.scala:79)
      1 frame
    6. Apache Kafka
      Processor.run
      1. kafka.message.MessageSet.foreach(MessageSet.scala:63)
      2. kafka.log.Log.append(Log.scala:185)
      3. kafka.server.KafkaRequestHandlers.handleProducerRequest(KafkaRequestHandlers.scala:56)
      4. kafka.server.KafkaRequestHandlers$$anonfun$handlerFor$1.apply(KafkaRequestHandlers.scala:40)
      5. kafka.server.KafkaRequestHandlers$$anonfun$handlerFor$1.apply(KafkaRequestHandlers.scala:40)
      6. kafka.network.Processor.handle(SocketServer.scala:268)
      7. kafka.network.Processor.read(SocketServer.scala:291)
      8. kafka.network.Processor.run(SocketServer.scala:202)
      8 frames
    7. Java RT
      Thread.run
      1. java.lang.Thread.run(Thread.java:662)
      1 frame