Update to version 3.0 client and update interfaces

opentracing-kafka-client/pom.xml

@@ -32,6 +32,11 @@
       <artifactId>kafka-clients</artifactId>
       <scope>provided</scope>
     </dependency>
+    <dependency>
+      <groupId>org.slf4j</groupId>
+      <artifactId>slf4j-api</artifactId>
+      <version>1.7.30</version>
+    </dependency>
   </dependencies>
 
 </project>

opentracing-kafka-client/src/main/java/io/opentracing/contrib/kafka/TracingKafkaConsumer.java

@@ -22,6 +22,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
+import java.util.OptionalLong;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiFunction;
@@ -224,7 +225,7 @@ public Map<TopicPartition, OffsetAndMetadata> committed(Set<TopicPartition> part
   public Map<TopicPartition, OffsetAndMetadata> committed(Set<TopicPartition> partitions,
       final Duration timeout) {
     return consumer.committed(partitions, timeout);
-  }
+ }
 
   @Override
   public Map<MetricName, ? extends Metric> metrics() {
@@ -300,6 +301,11 @@ public Map<TopicPartition, Long> endOffsets(Collection<TopicPartition> collectio
     return consumer.endOffsets(collection, duration);
   }
 
+  @Override
+  public OptionalLong currentLag(final TopicPartition topicPartition) {
+    return consumer.currentLag(topicPartition);
+  }
+
   @Override
   public ConsumerGroupMetadata groupMetadata() {
     return consumer.groupMetadata();
@@ -311,14 +317,13 @@ public void enforceRebalance() {
   }
 
   @Override
-  public void close() {
-    consumer.close();
+  public void enforceRebalance(final String s) {
+    consumer.enforceRebalance(s);
   }
 
   @Override
-  @Deprecated
-  public void close(long l, TimeUnit timeUnit) {
-    consumer.close(l, timeUnit);
+  public void close() {
+    consumer.close();
   }
 
   @Override

opentracing-kafka-client/src/main/java/io/opentracing/contrib/kafka/TracingKafkaProducer.java

@@ -166,9 +166,4 @@ public void close(Duration duration) {
     producer.close(duration);
   }
 
-  @Override
-  public void close(long timeout, TimeUnit timeUnit) {
-    producer.close(timeout, timeUnit);
-  }
-
 }

opentracing-kafka-spring/src/test/java/io/opentracing/contrib/kafka/spring/TracingSpringKafkaTest.java

@@ -60,9 +60,10 @@ public void test() {
     await().atMost(15, TimeUnit.SECONDS).until(reportedSpansSize(), greaterThanOrEqualTo(3));
 
     List<MockSpan> spans = mockTracer.finishedSpans();
+    spans.forEach(s -> System.out.println("Span Operation: " + s.operationName()));
     assertThat(spans, contains(
-        new SpanMatcher("To_spring"),
         new SpanMatcher("From_spring"),
+        new SpanMatcher("To_spring"),
         new SpanMatcher("KafkaListener_spring")));
   }
 

opentracing-kafka-streams/src/main/java/io/opentracing/contrib/kafka/streams/TracingKafkaClientSupplier.java

@@ -25,6 +25,8 @@
 import java.util.Collections;
 import java.util.Map;
 import java.util.function.BiFunction;
+
+import org.apache.kafka.clients.admin.Admin;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -95,6 +97,11 @@ public AdminClient getAdminClient(final Map<String, Object> config) {
     return AdminClient.create(config);
   }
 
+  @Override
+  public Admin getAdmin(final Map<String, Object> config) {
+    return this.getAdminClient(config);
+  }
+
   @Override
   public Producer<byte[], byte[]> getProducer(Map<String, Object> config) {
     return new TracingKafkaProducerBuilder<>(

pom.xml

@@ -62,13 +62,13 @@
   </issueManagement>
 
   <properties>
-    <java.version>1.8</java.version>
+    <java.version>11</java.version>
     <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     <project.build.resourceEncoding>UTF-8</project.build.resourceEncoding>
 
-    <opentracing.version>0.33.0</opentracing.version>
-    <kafka.version>2.6.0</kafka.version>
-    <spring.kafka.version>2.6.1</spring.kafka.version>
+    <opentracing.version>0.32.0</opentracing.version>
+    <kafka.version>3.2.1</kafka.version>
+    <spring.kafka.version>2.9.0</spring.kafka.version>
     <spring.version>5.2.7.RELEASE</spring.version>
     <coveralls-maven-plugin.version>4.3.0</coveralls-maven-plugin.version>
     <jacoco-maven-plugin.version>0.8.5</jacoco-maven-plugin.version>
@@ -128,7 +128,7 @@
         </exclusion>
         <exclusion>
           <groupId>org.apache.kafka</groupId>
-          <artifactId>kafka_2.11</artifactId>
+          <artifactId>kafka_2.12</artifactId>
         </exclusion>
         <exclusion>
           <groupId>org.apache.kafka</groupId>
@@ -268,7 +268,7 @@
           <plugin>
             <groupId>org.apache.maven.plugins</groupId>
             <artifactId>maven-source-plugin</artifactId>
-            <version>3.2.0</version>
+            <version>3.2.1</version>
             <executions>
               <execution>
                 <id>attach-sources</id>
@@ -282,7 +282,7 @@
           <plugin>
             <groupId>org.apache.maven.plugins</groupId>
             <artifactId>maven-javadoc-plugin</artifactId>
-            <version>3.1.1</version>
+            <version>3.4.1</version>
             <configuration>
               <failOnError>false</failOnError>
             </configuration>