Update to Kafka 1.0.1 in kafka-indexing

[sixtus]

The dependency is still on 0.10 and I don't see any argument for keeping it at that version.

I tried upgrading the version in pom.xml, it compiles, but the tests timeout. I currently don't have the time to debug it, so I thought I open a ticket and see if it's obvious to someone else.

[pdeva]

this is badly needed to ensure a fully exactly-once pipeline.
currently without the transactional topics introduced in kafka 1.0, the exactly-once is not guaranteed from the producer side.

i hope this gets earmarked for druid 0.13.0

[pdeva]

i believe # #5404 is related to this

[gianm]

Upgrading is good and I do think #5404 is related. If anyone is willing to tackle this then go for it. Otherwise someone here will likely do it at some point.

[jwroblewski]

I was also looking into this, but bumped kafka directly to 1.1.0. Indexer pulls data from my local broker just fine, but KafkaSupervisor tests fail with:

java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.NotLeaderForPartitionException: This server is not the leader for that topic-partition.
	at io.druid.indexing.kafka.supervisor.KafkaSupervisorTest.addSomeEvents(KafkaSupervisorTest.java:1915)

When I connect to the in-memory broker from outside I get the following:

$ kafkacat -L -b localhost:12345
Metadata for all topics (from broker 1: localhost:12345/1):
 1 brokers:
  broker 1 at localhost:12345
 35 topics:
  topic "testTopic26" with 3 partitions:
    partition 2, leader 1, replicas: 1, isrs: 1
    partition 1, leader 1, replicas: 1, isrs: 1
    partition 0, leader 1, replicas: 1, isrs: 1
  topic "testTopic33" with 3 partitions:
    partition 2, leader 1, replicas: 1, isrs: 1
    partition 1, leader 1, replicas: 1, isrs: 1
    partition 0, leader 1, replicas: 1, isrs: 1
  topic "testTopic0" with 3 partitions:
    partition 2, leader 1, replicas: 1, isrs: 1
    partition 1, leader 1, replicas: 1, isrs: 1
    partition 0, leader 1, replicas: 1, isrs: 1
(... and so on ...)

Console producer fails in the same way as tests:

$ ./kafka-console-producer.sh --broker-list localhost:12345 --topic abc
>test
[2018-04-19 11:37:31,803] WARN [Producer clientId=console-producer] Error while fetching metadata with correlation id 1 : {abc=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
[2018-04-19 11:37:31,912] WARN [Producer clientId=console-producer] Error while fetching metadata with correlation id 3 : {abc=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)
>[2018-04-19 11:37:33,075] WARN [Producer clientId=console-producer] Got error produce response with correlation id 5 on topic-partition abc-0, retrying (2 attempts left). Error: NOT_LEADER_FOR_PARTITION (org.apache.kafka.clients.producer.internals.Sender)
[2018-04-19 11:37:33,179] WARN [Producer clientId=console-producer] Got error produce response with correlation id 7 on topic-partition abc-0, retrying (1 attempts left). Error: NOT_LEADER_FOR_PARTITION (org.apache.kafka.clients.producer.internals.Sender)
[2018-04-19 11:37:33,285] WARN [Producer clientId=console-producer] Got error produce response with correlation id 9 on topic-partition abc-0, retrying (0 attempts left). Error: NOT_LEADER_FOR_PARTITION (org.apache.kafka.clients.producer.internals.Sender)
[2018-04-19 11:37:33,392] ERROR Error when sending message to topic abc with key: null, value: 3 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback)
org.apache.kafka.common.errors.NotLeaderForPartitionException: This server is not the leader for that topic-partition.

Funny enough, when I have test code talk to a 1.1.0 instance running directly on my box (so not managed by the test suite), everything is fine.

It looks like the broker instance used by test is somehow misconfigured. Copying all params from my local kafka's server.properties didn't help, though.

Btw, I also tried using the 1.2.0-SNAPSHOT kafka compiled locally, but the errors were still there.

[gianm]

@jwroblewski Interesting… did you try enabling debug logs and seeing if the Kafka broker is logging anything interesting? You should be able to do that by editing common/src/main/resources/log4j2.xml.

[jwroblewski]

@gianm There are indeed some kafka exceptions in the logs which my IDE was swallowing before, I had to configure additional file appender to capture them. 1.2.0-SNAPSHOT was used here, but the same happens for 1.1.0:

2018-04-20 13:57:45,598 ERROR k.s.KafkaApis [kafka-request-handler-1] [KafkaApi-1] Error when handling request {controller_id=1,controller_epoch=1,partition_states=[{topic=abc,partition=2,controller_epoch=1,leader=1,leader_epoch=0,isr=[1],zk_version=0,replicas=[1],is_new=true},{topic=abc,partition=0,controller_epoch=1,leader=1,leader_epoch=0,isr=[1],zk_version=0,replicas=[1],is_new=true},{topic=abc,partition=1,controller_epoch=1,leader=1,leader_epoch=0,isr=[1],zk_version=0,replicas=[1],is_new=true}],live_leaders=[{id=1,host=localhost,port=13736}]}
java.lang.NoClassDefFoundError: org/slf4j/event/Level
	at kafka.utils.CoreUtils$.swallow$default$3(CoreUtils.scala:83) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.log.AbstractIndex.<init>(AbstractIndex.scala:79) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.log.OffsetIndex.<init>(OffsetIndex.scala:53) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.log.LogSegment$.open(LogSegment.scala:558) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.log.Log.loadSegments(Log.scala:429) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.log.Log.<init>(Log.scala:233) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.log.Log$.apply(Log.scala:1776) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.log.LogManager$$anonfun$getOrCreateLog$1.apply(LogManager.scala:678) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.log.LogManager$$anonfun$getOrCreateLog$1.apply(LogManager.scala:646) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at scala.Option.getOrElse(Option.scala:121) ~[scala-library-2.11.12.jar:?]
	at kafka.log.LogManager.getOrCreateLog(LogManager.scala:646) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.cluster.Partition$$anonfun$getOrCreateReplica$1.apply(Partition.scala:177) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.cluster.Partition$$anonfun$getOrCreateReplica$1.apply(Partition.scala:173) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.utils.Pool.getAndMaybePut(Pool.scala:65) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.cluster.Partition.getOrCreateReplica(Partition.scala:172) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.cluster.Partition$$anonfun$6$$anonfun$8.apply(Partition.scala:259) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.cluster.Partition$$anonfun$6$$anonfun$8.apply(Partition.scala:259) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234) ~[scala-library-2.11.12.jar:?]
	at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234) ~[scala-library-2.11.12.jar:?]
	at scala.collection.Iterator$class.foreach(Iterator.scala:891) ~[scala-library-2.11.12.jar:?]
	at scala.collection.AbstractIterator.foreach(Iterator.scala:1334) ~[scala-library-2.11.12.jar:?]
	at scala.collection.IterableLike$class.foreach(IterableLike.scala:72) ~[scala-library-2.11.12.jar:?]
	at scala.collection.AbstractIterable.foreach(Iterable.scala:54) ~[scala-library-2.11.12.jar:?]
	at scala.collection.TraversableLike$class.map(TraversableLike.scala:234) ~[scala-library-2.11.12.jar:?]
	at scala.collection.AbstractTraversable.map(Traversable.scala:104) ~[scala-library-2.11.12.jar:?]
	at kafka.cluster.Partition$$anonfun$6.apply(Partition.scala:259) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.cluster.Partition$$anonfun$6.apply(Partition.scala:253) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.utils.CoreUtils$.inLock(CoreUtils.scala:250) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.utils.CoreUtils$.inWriteLock(CoreUtils.scala:258) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.cluster.Partition.makeLeader(Partition.scala:253) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.server.ReplicaManager$$anonfun$makeLeaders$4.apply(ReplicaManager.scala:1161) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.server.ReplicaManager$$anonfun$makeLeaders$4.apply(ReplicaManager.scala:1159) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:130) ~[scala-library-2.11.12.jar:?]
	at scala.collection.mutable.HashMap$$anonfun$foreach$1.apply(HashMap.scala:130) ~[scala-library-2.11.12.jar:?]
	at scala.collection.mutable.HashTable$class.foreachEntry(HashTable.scala:236) ~[scala-library-2.11.12.jar:?]
	at scala.collection.mutable.HashMap.foreachEntry(HashMap.scala:40) ~[scala-library-2.11.12.jar:?]
	at scala.collection.mutable.HashMap.foreach(HashMap.scala:130) ~[scala-library-2.11.12.jar:?]
	at kafka.server.ReplicaManager.makeLeaders(ReplicaManager.scala:1159) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.server.ReplicaManager.becomeLeaderOrFollower(ReplicaManager.scala:1079) ~[kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.server.KafkaApis.handleLeaderAndIsrRequest(KafkaApis.scala:183) [kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.server.KafkaApis.handle(KafkaApis.scala:108) [kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at kafka.server.KafkaRequestHandler.run(KafkaRequestHandler.scala:69) [kafka_2.11-1.2.0-SNAPSHOT.jar:?]
	at java.lang.Thread.run(Thread.java:748) [?:1.8.0_162]
Caused by: java.lang.ClassNotFoundException: org.slf4j.event.Level
	at java.net.URLClassLoader.findClass(URLClassLoader.java:381) ~[?:1.8.0_162]
	at java.lang.ClassLoader.loadClass(ClassLoader.java:424) ~[?:1.8.0_162]
	at java.lang.ClassLoader.loadClass(ClassLoader.java:357) ~[?:1.8.0_162]
	at org.apache.maven.surefire.booter.IsolatedClassLoader.loadClass(IsolatedClassLoader.java:97) ~[?:?]
	... 43 more

The missing org.slf4j.event.Level was introduced in slf4j-api 1.7.5, and druid is still at 1.6.4 plus indexer excludes the slf4j-api dependency from kafka artifact. After updating to 1.7.25 in the main pom.xml, tests are now passing!

The above also explains why I did not experience the same problem when running tests against an external kafka broker.

[jwroblewski]

See #5674.

[maver1ck]

Maybe 2.0.0 at this point of time ?

[stale]

This issue has been marked as stale due to 280 days of inactivity. It will be closed in 4 weeks if no further activity occurs. If this issue is still relevant, please simply write any comment. Even if closed, you can still revive the issue at any time or discuss it on the dev@druid.apache.org list. Thank you for your contributions.

[jihoonson]

This issue is already fixed in #6496.