Fixed TracingKafkaClientSupplier to properly override getAdmin (#1312)

* Fixed TracingKafkaClientSupplier to properly override getAdmin

getAdminClient has been deprecated in favor of getAdmin.
Calls to getAdminClient are throwing an UnsupportedOperationException in kafka-streams version < 3.0.0,
while calls to getAdmin are also throwing an UnsupportedOperationException in kafka-streams version >= 3.0.0

* Updated license header

* Added tests

* Added missing license header

instrumentation/kafka-streams/src/main/java/brave/kafka/streams/TracingKafkaClientSupplier.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2013-2019 The OpenZipkin Authors
+ * Copyright 2013-2021 The OpenZipkin Authors
  *
  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
  * in compliance with the License. You may obtain a copy of the License at
@@ -37,7 +37,11 @@ final class TracingKafkaClientSupplier implements KafkaClientSupplier {
     this.kafkaTracing = kafkaTracing;
   }
 
-  @Override public AdminClient getAdminClient(Map<String, Object> config) {
+  @Deprecated public AdminClient getAdminClient(Map<String, Object> config) {
+    return getAdmin(config);
+  }
+
+  @Override public AdminClient getAdmin(Map<String, Object> config) {
     return AdminClient.create(config);
   }
 

instrumentation/kafka-streams/src/test/java/brave/kafka/streams/TracingKafkaClientSupplierTests.java

@@ -0,0 +1,69 @@
+/*
+ * Copyright 2013-2021 The OpenZipkin Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
+ * in compliance with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License
+ * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
+ * or implied. See the License for the specific language governing permissions and limitations under
+ * the License.
+ */
+package brave.kafka.streams;
+
+import java.util.Collections;
+import java.util.Map;
+
+import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.producer.Producer;
+import org.apache.kafka.streams.KafkaClientSupplier;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+class TracingKafkaClientSupplierTests {
+
+  final Map<String, Object> props = Collections.singletonMap("bootstrap.servers","localhost:9092");
+
+  @Test
+  public void shouldReturnNewAdminClient() {
+    TracingKafkaClientSupplier supplier = new TracingKafkaClientSupplier(null);
+    assertThat(supplier.getAdminClient(props)).isNotNull();
+    assertThat(supplier.getAdmin(props)).isNotNull();
+  }
+
+  @Test
+  void shouldThrowException() {
+    assertThrows(UnsupportedOperationException.class, () -> {
+      FakeKafkaClientSupplier fake = new FakeKafkaClientSupplier();
+      fake.getAdminClient(props);
+    });
+  }
+
+  private static class FakeKafkaClientSupplier implements KafkaClientSupplier {
+
+    @Override
+    public Producer<byte[], byte[]> getProducer(Map<String, Object> map) {
+      return null;
+    }
+
+    @Override
+    public Consumer<byte[], byte[]> getConsumer(Map<String, Object> map) {
+      return null;
+    }
+
+    @Override
+    public Consumer<byte[], byte[]> getRestoreConsumer(Map<String, Object> map) {
+      return null;
+    }
+
+    @Override
+    public Consumer<byte[], byte[]> getGlobalConsumer(Map<String, Object> map) {
+      return null;
+    }
+  }
+
+}