simplify

Author: trask

instrumentation/kafka/kafka-clients/kafka-clients-2.6/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/v2_6/KafkaTelemetry.java

@@ -26,6 +26,7 @@
 import io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.OpenTelemetryMetricsReporter;
 import io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.OpenTelemetrySupplier;
 import io.opentelemetry.instrumentation.kafkaclients.common.v0_11.internal.TracingList;
+import io.opentelemetry.instrumentation.kafkaclients.v2_6.internal.KafkaTelemetrySupplier;
 import io.opentelemetry.instrumentation.kafkaclients.v2_6.internal.OpenTelemetryConsumerInterceptor;
 import io.opentelemetry.instrumentation.kafkaclients.v2_6.internal.OpenTelemetryProducerInterceptor;
 import java.lang.reflect.InvocationTargetException;

instrumentation/kafka/kafka-clients/kafka-clients-2.6/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/v2_6/KafkaTelemetrySupplier.java renamed to instrumentation/kafka/kafka-clients/kafka-clients-2.6/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/v2_6/internal/KafkaTelemetrySupplier.java

@@ -3,8 +3,9 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package io.opentelemetry.instrumentation.kafkaclients.v2_6;
+package io.opentelemetry.instrumentation.kafkaclients.v2_6.internal;
 
+import io.opentelemetry.instrumentation.kafkaclients.v2_6.KafkaTelemetry;
 import java.io.Serializable;
 import java.util.Objects;
 import java.util.function.Supplier;
@@ -16,17 +17,17 @@
  * <p>This class is internal and is hence not for public use. Its APIs are unstable and can change
  * at any time.
  */
-final class KafkaTelemetrySupplier implements Supplier<Object>, Serializable {
+public final class KafkaTelemetrySupplier implements Supplier<KafkaTelemetry>, Serializable {
   private static final long serialVersionUID = 1L;
-  private final transient Object kafkaTelemetry;
+  private final transient KafkaTelemetry kafkaTelemetry;
 
-  KafkaTelemetrySupplier(Object kafkaTelemetry) {
+  public KafkaTelemetrySupplier(KafkaTelemetry kafkaTelemetry) {
     Objects.requireNonNull(kafkaTelemetry);
     this.kafkaTelemetry = kafkaTelemetry;
   }
 
   @Override
-  public Object get() {
+  public KafkaTelemetry get() {
     return kafkaTelemetry;
   }
 

instrumentation/kafka/kafka-clients/kafka-clients-2.6/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/v2_6/internal/OpenTelemetryConsumerInterceptor.java

@@ -13,7 +13,6 @@
 import io.opentelemetry.instrumentation.kafkaclients.v2_6.KafkaTelemetry;
 import java.util.Map;
 import java.util.Objects;
-import java.util.function.Supplier;
 import javax.annotation.Nullable;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerInterceptor;
@@ -69,21 +68,21 @@ public void configure(Map<String, ?> configs) {
       return;
     }
 
-    if (!(telemetrySupplier instanceof Supplier)) {
-      throw new IllegalStateException(
-          "Configuration property "
-              + CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER
-              + " is not instance of Supplier");
-    }
+    KafkaTelemetrySupplier supplier =
+        getProperty(configs, CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER, KafkaTelemetrySupplier.class);
+    this.telemetry = supplier.get();
+  }
 
-    Object kafkaTelemetry = ((Supplier<?>) telemetrySupplier).get();
-    if (!(kafkaTelemetry instanceof KafkaTelemetry)) {
+  @SuppressWarnings("unchecked")
+  private static <T> T getProperty(Map<String, ?> configs, String key, Class<T> requiredType) {
+    Object value = configs.get(key);
+    if (value == null) {
+      throw new IllegalStateException("Missing required configuration property: " + key);
+    }
+    if (!requiredType.isInstance(value)) {
       throw new IllegalStateException(
-          "Configuration property "
-              + CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER
-              + " supplier does not return KafkaTelemetry instance");
+          "Configuration property " + key + " is not instance of " + requiredType.getSimpleName());
     }
-
-    this.telemetry = (KafkaTelemetry) kafkaTelemetry;
+    return (T) value;
   }
 }

instrumentation/kafka/kafka-clients/kafka-clients-2.6/library/src/main/java/io/opentelemetry/instrumentation/kafkaclients/v2_6/internal/OpenTelemetryProducerInterceptor.java

@@ -9,7 +9,6 @@
 import io.opentelemetry.instrumentation.kafkaclients.v2_6.KafkaTelemetry;
 import java.util.Map;
 import java.util.Objects;
-import java.util.function.Supplier;
 import javax.annotation.Nullable;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerInterceptor;
@@ -54,21 +53,21 @@ public void configure(Map<String, ?> configs) {
       return;
     }
 
-    if (!(telemetrySupplier instanceof Supplier)) {
-      throw new IllegalStateException(
-          "Configuration property "
-              + CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER
-              + " is not instance of Supplier");
-    }
+    KafkaTelemetrySupplier supplier =
+        getProperty(configs, CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER, KafkaTelemetrySupplier.class);
+    this.telemetry = supplier.get();
+  }
 
-    Object kafkaTelemetry = ((Supplier<?>) telemetrySupplier).get();
-    if (!(kafkaTelemetry instanceof KafkaTelemetry)) {
+  @SuppressWarnings("unchecked")
+  private static <T> T getProperty(Map<String, ?> configs, String key, Class<T> requiredType) {
+    Object value = configs.get(key);
+    if (value == null) {
+      throw new IllegalStateException("Missing required configuration property: " + key);
+    }
+    if (!requiredType.isInstance(value)) {
       throw new IllegalStateException(
-          "Configuration property "
-              + CONFIG_KEY_KAFKA_TELEMETRY_SUPPLIER
-              + " supplier does not return KafkaTelemetry instance");
+          "Configuration property " + key + " is not instance of " + requiredType.getSimpleName());
     }
-
-    this.telemetry = (KafkaTelemetry) kafkaTelemetry;
+    return (T) value;
   }
 }

instrumentation/kafka/kafka-clients/kafka-clients-2.6/library/src/test/java/io/opentelemetry/instrumentation/kafkaclients/v2_6/KafkaTelemetryInterceptorTest.java

@@ -8,6 +8,7 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
+import io.opentelemetry.instrumentation.kafkaclients.v2_6.internal.KafkaTelemetrySupplier;
 import io.opentelemetry.instrumentation.kafkaclients.v2_6.internal.OpenTelemetryConsumerInterceptor;
 import io.opentelemetry.instrumentation.kafkaclients.v2_6.internal.OpenTelemetryProducerInterceptor;
 import io.opentelemetry.instrumentation.testing.junit.InstrumentationExtension;