Federated Clients: implement various federated consumer methods #119

jonlee2 · 2019-05-17T00:28:05Z

No description provided.

...e-kafka-clients/src/main/java/com/linkedin/kafka/clients/consumer/LiKafkaConsumerConfig.java

radai-rosenblatt · 2019-05-20T20:31:26Z

...-clients/src/main/java/com/linkedin/kafka/clients/consumer/LiKafkaFederatedConsumerImpl.java

  }

  @Override
  public void commitAsync(OffsetCommitCallback callback) {
-    throw new UnsupportedOperationException("Not implemented yet");
+    for (ClusterConsumerPair<K, V> entry : getImmutableConsumerList()) {
+      entry.getConsumer().commitAsync(callback);


this would call a user-provided callback N times for N consumers. we need to aggregate them 1st and call the user callback only once

Thanks for catching this. Added a wrapper callback, which will call the user callback once.

radai-rosenblatt · 2019-05-20T20:31:50Z

...-clients/src/main/java/com/linkedin/kafka/clients/consumer/LiKafkaFederatedConsumerImpl.java

+    List<ClusterConsumerPair<K, V>> consumers = getImmutableConsumerList();
+    for (Map.Entry<ClusterDescriptor, Map<TopicPartition, OffsetAndMetadata>> entry :
+        getPartitionValueMapByCluster(offsets).entrySet()) {
+      getOrCreatePerClusterConsumer(consumers, entry.getKey()).commitAsync(entry.getValue(), callback);


similar to above - user-provided callbacks should be called once

radai-rosenblatt · 2019-05-20T20:32:28Z

...-clients/src/main/java/com/linkedin/kafka/clients/consumer/LiKafkaFederatedConsumerImpl.java

-    throw new UnsupportedOperationException("Not implemented yet");
+    List<ClusterConsumerPair<K, V>> consumers = getImmutableConsumerList();
+    for (Map.Entry<ClusterDescriptor, Collection<TopicPartition>> entry : getPartitionsByCluster(partitions).entrySet()) {
+      getOrCreatePerClusterConsumer(consumers, entry.getKey()).seekToBeginning(entry.getValue());


not sure if seek() should result in the creation of a consumer - i think only subscribe/assign should

The current implementation of config reload closes clients but does not recreate them. They will be recreated at the subsequent operation. If we keep this "lazy" reload behavior, seek() may need to create a consumer.

i dont know what the lazy approach gets us - we expect these clients to be active.

radai-rosenblatt · 2019-05-20T20:34:27Z

...-clients/src/main/java/com/linkedin/kafka/clients/consumer/LiKafkaFederatedConsumerImpl.java

@@ -534,27 +694,17 @@ public void uncaughtException(Thread t, Throwable e) {
    try {
      if (!countDownLatch.await(deadlineTimeMs - System.currentTimeMillis(), TimeUnit.MILLISECONDS)) {
        LiKafkaClientsUtils.dumpStacksForAllLiveThreads(threads);
-        throw new KafkaException("Fail to close all consumers for cluster group " + _clusterGroup + " in " +
-            timeout + " " + timeUnit);
+        throw new KafkaException("fail to perform " + methodName + " for cluster group " + _clusterGroup + " in " +


for bonus points could iterate over threads still alive and print their stacks

I think that's what the line above (LiKafkaClientsUtils.dumpStacksForAllLiveThreads(threads)) is already doing? Did you mean something else?

...-clients/src/main/java/com/linkedin/kafka/clients/consumer/LiKafkaFederatedConsumerImpl.java

jonlee2 requested a review from radai-rosenblatt May 17, 2019 00:28

jonlee2 force-pushed the federated-consumer-misc branch from 60fafdf to d9f853a Compare May 17, 2019 00:30

radai-rosenblatt reviewed May 20, 2019

View reviewed changes

...e-kafka-clients/src/main/java/com/linkedin/kafka/clients/consumer/LiKafkaConsumerConfig.java Outdated Show resolved Hide resolved

radai-rosenblatt reviewed May 20, 2019

View reviewed changes

jonlee2 force-pushed the federated-consumer-misc branch 2 times, most recently from d749b92 to c823088 Compare June 14, 2019 18:44

jonlee2 requested a review from kehuum June 14, 2019 18:47

kehuum reviewed Jun 17, 2019

View reviewed changes