Kafka Streams instrumentation not compatible with kafka-streams 3.0.0

[codependent]

Describe the Bug

A detailed description of this bug can be found on Stackoverflow: https://stackoverflow.com/questions/70382128/tracingkafkaclientsupplier-error-implementations-of-kafkaclientsupplier-should

Steps to Reproduce

Create a Spring Boot 2.6.1 application with Spring Cloud 2021.0.0, adding spring-cloud-sleuth to instrument it.

Expected Behaviour

The application should be instrumented correctly, not crashing at startup.

TracingKafkaClientSupplier should override this method from KafkaClientSupplier:

default Admin getAdmin(final Map<String, Object> config) {
        throw new UnsupportedOperationException("Implementations of KafkaClientSupplier should implement the getAdmin() method.");
}

However the current implementation returns AdminClient instead of Admin, thus, throwing the default method exception:

@Override public AdminClient getAdminClient(Map<String, Object> config) {
        return AdminClient.create(config);
}

[jcchavezs]

@codependent is this solved by https://github.com/openzipkin/brave/pull/1312?

[codependent]

@jcchavezs I don't think that'll solve it for kafka-streams 3.0.0. The commit leaves TracingKafkaClientSupplier with:

  @Deprecated public AdminClient getAdminClient(Map<String, Object> config) {
    return getAdmin(config);
  }

  @Override public AdminClient getAdmin(Map<String, Object> config) {
    return AdminClient.create(config);
  }

However, In 3.0.0 the signature that should be overriden is different from the last method (notice it should return Admin, not AdminClient:

public interface KafkaClientSupplier {
    default Admin getAdmin(Map<String, Object> config) {
        throw new UnsupportedOperationException("Implementations of KafkaClientSupplier should implement the getAdmin() method.");
    }

[jcchavezs]

Are you able to work out a PR?

[garyrussell]

AdminClient is a subclass of Admin so it's ok for an implementation to narrow the type returned.

[codependent]

AdminClient is a subclass of Admin so it's ok for an implementation to narrow the type returned.

You're right, I didn't realize AdminClient is a subclass of Admin. So @jcchavezs, the PR you mentioned will solve this issue.

Thanks for having a look at it.

[jcchavezs]

I wonder if we could add tests for both Kafka streams version?

On Thu, Dec 16, 2021, 21:57 Jose A. Iñigo @.***> wrote:

AdminClient is a subclass of Admin so it's ok for an implementation to narrow the type returned.

You're right, I didn't realize AdminClient is a subclass of Admin. So @jcchavezs https://github.com/jcchavezs, the PR you mentioned will solve this issue.

Thanks for having a look at it.

— Reply to this email directly, view it on GitHub https://github.com/openzipkin/brave/issues/1316#issuecomment-996188127, or unsubscribe https://github.com/notifications/unsubscribe-auth/AAXOYATNHCCQ7S2NLGA3TWDURJHDNANCNFSM5KGXYFVA . Triage notifications on the go with GitHub Mobile for iOS https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675 or Android https://play.google.com/store/apps/details?id=com.github.android&referrer=utm_campaign%3Dnotification-email%26utm_medium%3Demail%26utm_source%3Dgithub.

You are receiving this because you were mentioned.Message ID: @.***>

[codependent]

Hi @jcchavezs, although @TYsewyn solved the problem originally described in this issue, I've come across another incompatibility between Brave and Kafka Streams 3.x:

java.lang.AbstractMethodError: Receiver class brave.kafka.clients.TracingConsumer does not define or inherit an implementation of the resolved method 'abstract java.util.OptionalLong currentLag(org.apache.kafka.common.TopicPartition)' of interface org.apache.kafka.clients.consumer.Consumer.
    at org.apache.kafka.streams.processor.internals.PartitionGroup.readyToProcess(PartitionGroup.java:143)

As you can see, TracingConsumer should implement OptionalLong currentLag(TopicPartition topicPartition);

[jcchavezs]

I think we should seriously consider a test with both <3.0.0 and post 3.0.0 to make sure everything works. For the moment could you fix this issue? Or even better come up with the test?

On Wed, Dec 22, 2021, 17:57 Jose A. Iñigo @.***> wrote:

Hi @jcchavezs https://github.com/jcchavezs, although @TYsewyn https://github.com/TYsewyn solved the problem originally described in this issue, I've come across another incompatibility between Brave and Kafka Streams 3.x:

java.lang.AbstractMethodError: Receiver class brave.kafka.clients.TracingConsumer does not define or inherit an implementation of the resolved method 'abstract java.util.OptionalLong currentLag(org.apache.kafka.common.TopicPartition)' of interface org.apache.kafka.clients.consumer.Consumer. at org.apache.kafka.streams.processor.internals.PartitionGroup.readyToProcess(PartitionGroup.java:143)

As you can see, TracingConsumer should implement OptionalLong currentLag(TopicPartition topicPartition);

[image: imagen] https://user-images.githubusercontent.com/6399869/147127915-17e88e1f-f8d5-4919-8677-32e13c26b1df.png

— Reply to this email directly, view it on GitHub https://github.com/openzipkin/brave/issues/1316#issuecomment-999726168, or unsubscribe https://github.com/notifications/unsubscribe-auth/AAXOYAVHW2K7UEGK4EMCHQDUSH7PXANCNFSM5KGXYFVA . Triage notifications on the go with GitHub Mobile for iOS https://apps.apple.com/app/apple-store/id1477376905?ct=notification-email&mt=8&pt=524675 or Android https://play.google.com/store/apps/details?id=com.github.android&referrer=utm_campaign%3Dnotification-email%26utm_medium%3Demail%26utm_source%3Dgithub.

You are receiving this because you were mentioned.Message ID: @.***>

[codependent]

Sure, I'll give it a try

[jeqo]

I wonder if we could keep the same approach as with v2.x, implementing latest methods without adding @Override annotation: e.g.

  // Do not use @Override annotation to avoid compatibility on deprecated methods
  public void close(long timeout, TimeUnit unit) {
    delegate.close(Duration.ofMillis(unit.convert(timeout, TimeUnit.MILLISECONDS)));
  }

The main consideration with v3.0 is that it removes KafkaConsumer#close(long, TimeUnit), keeping only KafkaConsumer#close(Duration) that is available since v2.0.

We could either break compatibility with <2.0 and keep the same module, or if we want to keep compatibility with <2.0, we may need to create a module for 3.0+.

wdyt?

[TYsewyn]

If the only problem with <2.0 support is the removal of KafkaConsumer#close(long, TimeUnit) I'd leave it like that and add the 3.x methods without the @Override annotation. In case there are more issues with keeping compatibility with <2.0, I would drop the support IMHO. The latest release is from July 2018. I don't see the added benefit of creating a separate module for 3.x in both cases. The @Override annotation gives only a compile-time warning, and won't cause any runtime effect.

Additionally, I have integration tests running locally that are passing for <2.0, 2.x and 3.x, for both kafka-clients and kafka-streams. The only problem is that I had to build a local version of kafka-junit since it doesn't support 3.0.0 yet.

[jeqo]

Managed to update kafka-junit to 3.0. #1322 is green now. @TYsewyn if you have some bandwidth to take a look would be much appreciated.

[kurbanek-worldremit]

@TYsewyn and @jeqo Could you move #1320 (with any dependencies) forward?

Currently dealing with:

java.lang.AbstractMethodError: Receiver class brave.kafka.clients.TracingConsumer does not define or inherit an implementation of the resolved method 'abstract java.util.OptionalLong currentLag(org.apache.kafka.common.TopicPartition)' of interface org.apache.kafka.clients.consumer.Consumer. at org.apache.kafka.streams.processor.internals.PartitionGroup.readyToProcess(PartitionGroup.java:143) at org.apache.kafka.streams.processor.internals.StreamTask.isProcessable(StreamTask.java:674) at org.apache.kafka.streams.processor.internals.StreamTask.process(StreamTask.java:694) at org.apache.kafka.streams.processor.internals.TaskManager.process(TaskManager.java:1193) at org.apache.kafka.streams.processor.internals.StreamThread.runOnce(StreamThread.java:753) at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:583) at org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:555)

[adamsz-wr]

up

[art-licis]

It seems that two related PRs #1322 and #1320 are stuck in a review phase for quite a long time - I'm sure this affects large userbase (latest Spring Boot with Kafka Streams), so giving it a little nudge to revive this thread.

[jeqo]

Brave 5.13.8 released https://github.com/openzipkin/brave/releases/tag/5.13.8

[fathimaSheikh]

@jeqo I'm using 5.13.8, `

io.zipkin.brave

<artifactId>brave</artifactId>
<version>5.13.8</version>

` but I still have the same issue.

java.lang.AbstractMethodError: Receiver class brave.kafka.clients.TracingConsumer does not define or inherit an implementation of the resolved method 'abstract java.util.OptionalLong currentLag(org.apache.kafka.common.TopicPartition)' of interface org.apache.kafka.clients.consumer.Consumer.

The code base https://github.com/openzipkin/brave/blob/master/instrumentation/kafka-clients/src/main/java/brave/kafka/clients/TracingConsumer.java does not have the changes needed to resolve the issue as the PR https://github.com/openzipkin/brave/pull/1320 was not merged

Is there any workaround? or am I missing something?

I am using Spring Boot 2.6 and spring-cloud-sleuth version 3.1.1 which seems to using 5.13.7 version of io.zipkin.brave

I have overridden now to use latest version of 5.13.8 of io.zipkin.brave

[fathimaSheikh]

@jeqo I'm using 5.13.8, but I still have the same issue.

java.lang.AbstractMethodError: Receiver class brave.kafka.clients.TracingConsumer does not define or inherit an implementation of the resolved method 'abstract java.util.OptionalLong currentLag(org.apache.kafka.common.TopicPartition)' of interface org.apache.kafka.clients.consumer.Consumer.

The code base https://github.com/openzipkin/brave/blob/master/instrumentation/kafka-clients/src/main/java/brave/kafka/clients/TracingConsumer.java does not have the changes needed to resolve the issue as the PR #1320 was not merged

Is there any workaround? or am I missing something?

I am using Spring Boot 2.6 and spring-cloud-sleuth version 3.1.1 which seems to using 5.13.7 version of io.zipkin.brave

I have overridden now to use latest version of 5.13.8 of io.zipkin.brave

Issue resolved by using 5.13.8 version of brave-instrumentation-kafka-clients `

io.zipkin.brave

  <artifactId>brave-instrumentation-kafka-clients</artifactId>
  <version>5.13.8</version>
</dependency>`

[art-licis]

For extra context - I also needed to bump io.zipkin.brave:brave-instrumentation-kafka-streams to 5.13.8.