Allow to bind KafkaClientMetrics for each decaton consumer

benchmark/src/main/java/com/linecorp/decaton/benchmark/DecatonRunner.java

@@ -78,7 +78,7 @@ public void init(Config config, Recording recording, ResourceTracker resourceTra
         subscription = SubscriptionBuilder
                 .newBuilder("decaton-benchmark")
                 .consumerConfig(props)
-                .properties(StaticPropertySupplier.of(properties))
+                .addProperties(StaticPropertySupplier.of(properties))
                 .processorsBuilder(
                         ProcessorsBuilder.consuming(config.topic(),
                                                     (TaskExtractor<Task>) bytes -> {

processor/src/it/java/com/linecorp/decaton/processor/CoreFunctionalityTest.java

@@ -49,7 +49,7 @@ public class CoreFunctionalityTest {
     public RandomRule randomRule = new RandomRule();
 
     @Test(timeout = 30000)
-    public void testProcessConcurrent() {
+    public void testProcessConcurrent() throws Exception {
         Random rand = randomRule.random();
         ProcessorTestSuite
                 .builder(rule)
@@ -65,7 +65,7 @@ public void testProcessConcurrent() {
     }
 
     @Test(timeout = 30000)
-    public void testProcessConcurrent_PartitionScopeProcessor() {
+    public void testProcessConcurrent_PartitionScopeProcessor() throws Exception {
         Random rand = randomRule.random();
         ProcessorTestSuite
                 .builder(rule)
@@ -80,7 +80,7 @@ public void testProcessConcurrent_PartitionScopeProcessor() {
     }
 
     @Test(timeout = 30000)
-    public void testProcessConcurrent_ThreadScopeProcessor() {
+    public void testProcessConcurrent_ThreadScopeProcessor() throws Exception {
         Random rand = randomRule.random();
         ProcessorTestSuite
                 .builder(rule)
@@ -95,7 +95,7 @@ public void testProcessConcurrent_ThreadScopeProcessor() {
     }
 
     @Test(timeout = 30000)
-    public void testAsyncTaskCompletion() {
+    public void testAsyncTaskCompletion() throws Exception {
         ExecutorService executorService = Executors.newFixedThreadPool(16);
         Random rand = randomRule.random();
         ProcessorTestSuite
@@ -126,7 +126,7 @@ public void testAsyncTaskCompletion() {
      * heap pressure by holding entire ProcessingContext instance.
      */
     @Test(timeout = 30000)
-    public void testGetCompletionInstanceLater() {
+    public void testGetCompletionInstanceLater() throws Exception {
         ExecutorService executorService = Executors.newFixedThreadPool(16);
         Random rand = randomRule.random();
         ProcessorTestSuite
@@ -149,7 +149,7 @@ public void testGetCompletionInstanceLater() {
     }
 
     @Test(timeout = 60000)
-    public void testSingleThreadProcessing() {
+    public void testSingleThreadProcessing() throws Exception {
         // Note that this processing semantics is not be considered as Decaton specification which users can rely on.
         // Rather, this is just a expected behavior based on current implementation when we set concurrency to 1.
         ProcessingGuarantee noDuplicates = new ProcessingGuarantee() {
@@ -189,7 +189,7 @@ public void doAssert() {
     }
 
     @Test(timeout = 30000)
-    public void testAsyncCompletionWithLeakAndTimeout() {
+    public void testAsyncCompletionWithLeakAndTimeout() throws Exception {
         Random rand = randomRule.random();
         ProcessorTestSuite
                 .builder(rule)

processor/src/it/java/com/linecorp/decaton/processor/RateLimiterTest.java

@@ -77,7 +77,7 @@ public void testPropertyDynamicSwitch() throws Exception {
                                                              .consuming(topicName,
                                                                         new ProtocolBuffersDeserializer<>(HelloTask.parser()))
                                                              .thenProcess(processor))
-                                  .properties(StaticPropertySupplier.of(rateProp)));
+                                  .addProperties(StaticPropertySupplier.of(rateProp)));
              DecatonClient<HelloTask> client = TestUtils.client(topicName, rule.bootstrapServers())) {
 
             int count = 0;

processor/src/it/java/com/linecorp/decaton/processor/RetryQueueingTest.java

@@ -221,7 +221,7 @@ public void testRetryQueuingExtractingWithDefaultMeta() throws Exception {
     }
 
     @Test(timeout = 60000)
-    public void testRetryQueueingFromCompletionTimeoutCallback() {
+    public void testRetryQueueingFromCompletionTimeoutCallback() throws Exception {
         ProcessorTestSuite
                 .builder(rule)
                 .numTasks(100)

processor/src/it/java/com/linecorp/decaton/processor/SubscriptionStateTest.java

@@ -39,7 +39,7 @@ public class SubscriptionStateTest {
     public static KafkaClusterRule rule = new KafkaClusterRule();
 
     @Test(timeout = 30000)
-    public void testStateTransition() {
+    public void testStateTransition() throws Exception {
         Map<Integer, List<State>> subscriptionStates = new HashMap<>();
         ProcessorTestSuite
                 .builder(rule)

processor/src/it/java/com/linecorp/decaton/processor/metrics/MetricsTest.java

@@ -37,10 +37,10 @@
 import com.linecorp.decaton.client.kafka.ProtocolBuffersKafkaSerializer;
 import com.linecorp.decaton.processor.DecatonProcessor;
 import com.linecorp.decaton.processor.runtime.ProcessorProperties;
+import com.linecorp.decaton.processor.runtime.ProcessorSubscription;
 import com.linecorp.decaton.processor.runtime.ProcessorsBuilder;
 import com.linecorp.decaton.processor.runtime.Property;
 import com.linecorp.decaton.processor.runtime.StaticPropertySupplier;
-import com.linecorp.decaton.processor.runtime.ProcessorSubscription;
 import com.linecorp.decaton.protobuf.ProtocolBuffersDeserializer;
 import com.linecorp.decaton.protocol.Decaton.DecatonTaskRequest;
 import com.linecorp.decaton.protocol.Sample.HelloTask;
@@ -111,7 +111,7 @@ public void testDetectStuckPartitions() throws Exception {
                                                              .consuming(topicName,
                                                                         new ProtocolBuffersDeserializer<>(HelloTask.parser()))
                                                              .thenProcess(processor))
-                                  .properties(StaticPropertySupplier.of(
+                                  .addProperties(StaticPropertySupplier.of(
                                           Property.ofStatic(ProcessorProperties.CONFIG_PARTITION_CONCURRENCY, 1),
                                           Property.ofStatic(ProcessorProperties.CONFIG_MAX_PENDING_RECORDS, 1))));
              Producer<String, DecatonTaskRequest> producer =

processor/src/main/java/com/linecorp/decaton/processor/metrics/Metrics.java

@@ -25,6 +25,8 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.Supplier;
 
+import org.apache.kafka.clients.consumer.Consumer;
+
 import com.linecorp.decaton.processor.metrics.internal.AvailableTags;
 
 import io.micrometer.core.instrument.Counter;
@@ -33,6 +35,7 @@
 import io.micrometer.core.instrument.Meter.Id;
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.core.instrument.Timer;
+import io.micrometer.core.instrument.binder.kafka.KafkaClientMetrics;
 import io.micrometer.core.instrument.composite.CompositeMeterRegistry;
 import io.micrometer.core.instrument.config.MeterFilter;
 
@@ -99,6 +102,20 @@ public void close() {
     }
 
     public class SubscriptionMetrics extends AbstractMetrics {
+        volatile KafkaClientMetrics kafkaClientMetrics;
+
+        public void bindClientMetrics(Consumer<String, byte[]> consumer) {
+            kafkaClientMetrics = new KafkaClientMetrics(consumer, availableTags.subscriptionScope());
+            kafkaClientMetrics.bindTo(registry);
+        }
+
+        private void closeClientMetrics() {
+            if (kafkaClientMetrics != null) {
+                kafkaClientMetrics.close();
+                kafkaClientMetrics = null;
+            }
+        }
+
         private Timer processDuration(String section) {
             return meter(() -> Timer.builder("subscription.process.durations")
                                     .description(String.format(
@@ -118,6 +135,12 @@ private Timer processDuration(String section) {
         public final Timer handlePausesTime = processDuration("pause");
 
         public final Timer commitOffsetTime = processDuration("commit");
+
+        @Override
+        public void close() {
+            super.close();
+            closeClientMetrics();
+        }
     }
 
     public class TaskMetrics extends AbstractMetrics {

processor/src/main/java/com/linecorp/decaton/processor/runtime/ProcessorProperties.java

@@ -138,6 +138,16 @@ public class ProcessorProperties extends AbstractDecatonProperties {
     public static final PropertyDefinition<Boolean> CONFIG_LOGGING_MDC_ENABLED =
             PropertyDefinition.define("decaton.logging.mdc.enabled", Boolean.class, true,
                                       v -> v instanceof Boolean);
+    /**
+     * Controls whether to enable or disable binding Micrometer's KafkaClientMetrics to decaton consumers.
+     * This is disabled for backwards compatiblity, but recommended if you rely on Micrometer
+     * since JMX metrics are deprecated. The downside is a possible increase in metrics count.
+     *
+     * Reloadable: no
+     */
+    public static final PropertyDefinition<Boolean> CONFIG_BIND_CLIENT_METRICS =
+            PropertyDefinition.define("decaton.client.metrics.micrometer.bound", Boolean.class, false,
+                                      v -> v instanceof Boolean);
     /**
      * Control time to "timeout" a deferred completion.
      * Decaton allows {@link DecatonProcessor}s to defer completion of a task by calling
@@ -174,6 +184,7 @@ public class ProcessorProperties extends AbstractDecatonProperties {
                     CONFIG_GROUP_REBALANCE_TIMEOUT_MS,
                     CONFIG_SHUTDOWN_TIMEOUT_MS,
                     CONFIG_LOGGING_MDC_ENABLED,
+                    CONFIG_BIND_CLIENT_METRICS,
                     CONFIG_DEFERRED_COMPLETE_TIMEOUT_MS));
 
     /**

processor/src/main/java/com/linecorp/decaton/processor/runtime/ProcessorSubscription.java

@@ -16,6 +16,8 @@
 
 package com.linecorp.decaton.processor.runtime;
 
+import static com.linecorp.decaton.processor.runtime.ProcessorProperties.CONFIG_BIND_CLIENT_METRICS;
+
 import java.time.Duration;
 import java.util.Collection;
 import java.util.Optional;
@@ -140,6 +142,9 @@ public void receive(ConsumerRecord<String, byte[]> record) {
         metrics = Metrics.withTags("subscription", scope.subscriptionId()).new SubscriptionMetrics();
 
         Consumer<String, byte[]> consumer = consumerSupplier.get();
+        if (props.get(CONFIG_BIND_CLIENT_METRICS).value()) {
+            metrics.bindClientMetrics(consumer);
+        }
         consumeManager = new ConsumeManager(consumer, contexts, new Handler(), metrics);
         commitManager = new CommitManager(
                 consumer, props.get(ProcessorProperties.CONFIG_COMMIT_INTERVAL_MS), contexts);

processor/src/main/java/com/linecorp/decaton/processor/runtime/SubscriptionBuilder.java

@@ -110,15 +110,22 @@ public static SubscriptionBuilder newBuilder(String subscriptionId) {
      * @param suppliers {@link PropertySupplier} instances
      * @return updated instance of {@link SubscriptionBuilder}.
      */
-    public SubscriptionBuilder properties(PropertySupplier... suppliers) {
-        ProcessorProperties.Builder<ProcessorProperties> builder = ProcessorProperties.builder();
+    public SubscriptionBuilder addProperties(PropertySupplier... suppliers) {
         for (PropertySupplier supplier : suppliers) {
-            builder.setBySupplier(supplier);
+            propertiesBuilder.setBySupplier(supplier);
         }
-        propertiesBuilder = builder;
         return this;
     }
 
+    /**
+     * Resets previous invocations of {@link #properties(PropertySupplier...)}
+     * and applies {@link #addProperties(PropertySupplier...)}.
+     */
+    public SubscriptionBuilder properties(PropertySupplier... suppliers) {
+        propertiesBuilder = ProcessorProperties.builder();
+        return addProperties(suppliers);
+    }
+
     /**
      * Configure subscription to enable retry processing when {@link DecatonProcessor} requests.
      *

testing/src/main/java/com/linecorp/decaton/testing/TestUtils.java

@@ -16,15 +16,25 @@
 
 package com.linecorp.decaton.testing;
 
+import static com.linecorp.decaton.processor.runtime.ProcessorProperties.CONFIG_BIND_CLIENT_METRICS;
+
+import java.time.Duration;
 import java.util.Optional;
 import java.util.Properties;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.LockSupport;
+import java.util.function.BooleanSupplier;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 
+import com.linecorp.decaton.processor.runtime.ProcessorProperties;
+import com.linecorp.decaton.processor.runtime.Property;
+import com.linecorp.decaton.processor.runtime.PropertySupplier;
+import com.linecorp.decaton.processor.runtime.StaticPropertySupplier;
 import com.linecorp.decaton.processor.runtime.SubscriptionStateListener;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.producer.KafkaProducer;
@@ -62,6 +72,7 @@ private static Properties defaultProducerProps(String bootstrapServers) {
     }
 
     public static final String DEFAULT_GROUP_ID = "test-group";
+    public static final Duration DEFINITELY_TOO_SLOW = Duration.ofSeconds(20);
 
     /**
      * A helper to instantiate {@link DecatonClient} for producing protobuf tasks with preset configurations
@@ -116,7 +127,8 @@ public static Producer<String, DecatonTaskRequest> producer(String bootstrapServ
      * @return {@link ProcessorSubscription} instance which is already running with unique subscription id assigned
      */
     public static ProcessorSubscription subscription(String bootstrapServers,
-                                                     Consumer<SubscriptionBuilder> builderConfigurer) {
+                                                     Consumer<SubscriptionBuilder> builderConfigurer)
+            throws InterruptedException, TimeoutException {
         return subscription("subscription-" + sequence(),
                             bootstrapServers,
                             builderConfigurer);
@@ -133,7 +145,8 @@ public static ProcessorSubscription subscription(String bootstrapServers,
      */
     public static ProcessorSubscription subscription(String subscriptionId,
                                                      String bootstrapServers,
-                                                     Consumer<SubscriptionBuilder> builderConfigurer) {
+                                                     Consumer<SubscriptionBuilder> builderConfigurer)
+            throws InterruptedException, TimeoutException {
         AtomicReference<SubscriptionStateListener> stateListenerRef = new AtomicReference<>();
         CountDownLatch initializationLatch = new CountDownLatch(1);
         SubscriptionStateListener outerStateListener = state -> {
@@ -160,14 +173,12 @@ public SubscriptionBuilder stateListener(SubscriptionStateListener stateListener
 
         builderConfigurer.accept(builder);
         builder.consumerConfig(props)
+               .addProperties(StaticPropertySupplier.of(Property.ofStatic(CONFIG_BIND_CLIENT_METRICS, true)))
                .stateListener(outerStateListener);
         ProcessorSubscription subscription = builder.buildAndStart();
 
-        try {
-            initializationLatch.await();
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-            throw new RuntimeException(e);
+        if (!initializationLatch.await(DEFINITELY_TOO_SLOW.toMillis(), TimeUnit.MILLISECONDS)) {
+            throw new TimeoutException("Initialization did not complete within " + DEFINITELY_TOO_SLOW);
         }
         return subscription;
     }
@@ -178,7 +189,7 @@ public SubscriptionBuilder stateListener(SubscriptionStateListener stateListener
      * @param condition expected condition to be met
      */
     public static void awaitCondition(String message,
-                                      Supplier<Boolean> condition) {
+                                      BooleanSupplier condition) {
         awaitCondition(message, condition, Long.MAX_VALUE);
     }
 
@@ -189,20 +200,15 @@ public static void awaitCondition(String message,
      * @param timeoutMillis max duration to wait
      */
     public static void awaitCondition(String message,
-                                      Supplier<Boolean> condition,
+                                      BooleanSupplier condition,
                                       long timeoutMillis) {
         long start = System.nanoTime();
-        while (!condition.get()) {
+        while (!condition.getAsBoolean()) {
             long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
             if (elapsedMillis >= timeoutMillis) {
                 throw new AssertionError(message);
             }
-            try {
-                Thread.sleep(100L);
-            } catch (InterruptedException e) {
-                Thread.currentThread().interrupt();
-                throw new RuntimeException(e);
-            }
+            LockSupport.parkNanos(TimeUnit.MILLISECONDS.toNanos(100));
         }
     }
 }