New KafkaConsumer failing

[chris-zen]

Hi @manub, I am having trouble to consume with the new kafka consumer. I see in your tests that you still use the old consumer for them. Have you experimented with the new one already ?

FYI: Some interesting things I saw in the logs:

ERROR kafka.server.KafkaApis: [KafkaApi-0] error when handling request null
java.lang.NoClassDefFoundError: org/apache/kafka/common/requests/JoinGroupRequest$GroupProtocol
    at kafka.server.KafkaApis$$anonfun$36.apply(KafkaApis.scala:791)
    at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
    at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
    at scala.collection.Iterator$class.foreach(Iterator.scala:742)
    at scala.collection.AbstractIterator.foreach(Iterator.scala:1194)
    at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
    at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
    at scala.collection.TraversableLike$class.map(TraversableLike.scala:245)
    at scala.collection.AbstractTraversable.map(Traversable.scala:104)
    at kafka.server.KafkaApis.handleJoinGroupRequest(KafkaApis.scala:791)
    at kafka.server.KafkaApis.handle(KafkaApis.scala:79)
    at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:60)
    at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.ClassNotFoundException: org.apache.kafka.common.requests.JoinGroupRequest$GroupProtocol
    at java.net.URLClassLoader.findClass(URLClassLoader.java:381)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:424)
    at sun.misc.Launcher$AppClassLoader.loadClass(Launcher.java:331)
    at java.lang.ClassLoader.loadClass(ClassLoader.java:357)
    ... 13 more
INFO  kafka.server.KafkaServer: [Kafka Server 0], Starting controlled shutdown

...

Unexpected error in join group response: The server experienced an unexpected error when processing the request
org.apache.kafka.common.KafkaException: Unexpected error in join group response: The server experienced an unexpected error when processing the request
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$JoinGroupResponseHandler.handle(AbstractCoordinator.java:376)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$JoinGroupResponseHandler.handle(AbstractCoordinator.java:324)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:665)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator$CoordinatorResponseHandler.onSuccess(AbstractCoordinator.java:644)
    at org.apache.kafka.clients.consumer.internals.RequestFuture$1.onSuccess(RequestFuture.java:167)
    at org.apache.kafka.clients.consumer.internals.RequestFuture.fireSuccess(RequestFuture.java:133)
    at org.apache.kafka.clients.consumer.internals.RequestFuture.complete(RequestFuture.java:107)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient$RequestFutureCompletionHandler.onComplete(ConsumerNetworkClient.java:380)
    at org.apache.kafka.clients.NetworkClient.poll(NetworkClient.java:274)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.clientPoll(ConsumerNetworkClient.java:320)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:213)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:193)
    at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:163)
    at org.apache.kafka.clients.consumer.internals.AbstractCoordinator.ensureActiveGroup(AbstractCoordinator.java:222)
    at org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.ensurePartitionAssignment(ConsumerCoordinator.java:311)
    at org.apache.kafka.clients.consumer.KafkaConsumer.pollOnce(KafkaConsumer.java:890)
    at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:853)

[chris-zen]

Sorry, forgot to mention that I am using the following versions:

"org.apache.kafka"            % "kafka-clients"                % "0.9.0.1"
"net.manub"                  %% "scalatest-embedded-kafka"     % "0.5.0" % "test"

[manub]

I've pushed the changes using the new consumer and the build works on my local machine (I'm waiting for a codeship build though).

One thing I noticed is that in order to throw a KafkaUnavailableException instead of relying on errors while creating the consumer I have to force a timeout asking for partitions of a topic, as it seems to me poll() hangs when there's no Kafka running, but you can still create the consumer without any problems.

Let me know if the changes in the codebase are helpful - I actually haven't seen at all the classpath issue you mentioned.

[chris-zen]

Hi @manub, I really appreciate your fast response and update.

After looking at your code and seeing that was pretty similar to mine, I went to run sbt test and then everything worked perfectly. Apparently the problem was in running the tests from within IntelliJ, because it was not understanding my dependency definition from build.sbt that looked like:

"net.manub"                  %% "scalatest-embedded-kafka"     % "0.5.0" % "test,it"

The moment I replace "test,it" by "test" it worked too.

So your embedded kafka works perfect with 0.9 consumer. Thank you very much 😄

By the way, I am not sure I understood the need to call partitionsFor(topic) before the poll.

[manub]

@chris-zen that's because for me consumer.poll() hangs when there's no Kafka running. calling partitionsFor() instead throws a timeout after 20 seconds. I had my build frozen because of that. I'd be happy to check for kafka running in a neater way.