(kafka) branch trunk updated: KAFKA-19697 NPE Cannot invoke org.apache.kafka.connect.runtime.ConnectMetrics$MetricGroup.close() (#21717)

[chia7712]

This is an automated email from the ASF dual-hosted git repository.

chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git


The following commit(s) were added to refs/heads/trunk by this push:
     new eb0a7358c45 KAFKA-19697 NPE Cannot invoke 
org.apache.kafka.connect.runtime.ConnectMetrics$MetricGroup.close() (#21717)
eb0a7358c45 is described below

commit eb0a7358c453782f07c860fe94f8e5e1c8ef7f6f
Author: Ken Huang <s7133...@gmail.com>
AuthorDate: Mon Mar 16 23:33:54 2026 +0800

    KAFKA-19697 NPE Cannot invoke 
org.apache.kafka.connect.runtime.ConnectMetrics$MetricGroup.close() (#21717)
    
    The PR fixed an NPE when connectorStatusMetrics unregistering task. The
    issue will happen if a connector has more than 1 task on one worker and
    these tasks fail to start. For example, if the DNS cant resolve producer
    URL, tasks will fail to build. If these tasks failing to start,
    ConnectorStatusMetricsGroup will unregister them. After first task
    unregistered , connectorStatusMetrics removed connector. When second
    task calls `recordTaskRemoved()`, connectorStatusMetrics is empty, so
    `connectorStatusMetrics.get(connectorTaskId.connector()).close();` will
    throw NPE.  Like in method `recordTaskAdded()`, this fix checks
    connectorStatusMetrics at the beginning of method `recordTaskRemoved()`.
    
    Co-authored-by: Fan Yang <fan...@hotmail.com>
    
    Reviewers: Ding <isdin...@163.com>, Chia-Ping Tsai <chia7...@gmail.com>
---
 .../org/apache/kafka/connect/runtime/Worker.java   |  4 +--
 .../apache/kafka/connect/runtime/WorkerTest.java   | 30 ++++++++++++++++++++++
 2 files changed, 32 insertions(+), 2 deletions(-)

diff --git 
a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java 
b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java
index 076953268d6..adc9367085f 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java
@@ -2396,8 +2396,8 @@ public final class Worker {
         protected synchronized void recordTaskRemoved(ConnectorTaskId 
connectorTaskId) {
             // Unregister connector task count metric if we remove the last 
task of the connector
             if (tasks.keySet().stream().noneMatch(id -> 
id.connector().equals(connectorTaskId.connector()))) {
-                
connectorStatusMetrics.get(connectorTaskId.connector()).close();
-                connectorStatusMetrics.remove(connectorTaskId.connector());
+                MetricGroup metricGroup = 
connectorStatusMetrics.remove(connectorTaskId.connector());
+                if (metricGroup != null) metricGroup.close();
             }
         }
 
diff --git 
a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java
 
b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java
index b4e83b76822..208c7154875 100644
--- 
a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java
+++ 
b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java
@@ -146,6 +146,7 @@ import static 
org.apache.kafka.connect.runtime.distributed.DistributedConfig.GRO
 import static 
org.apache.kafka.connect.runtime.distributed.DistributedConfig.OFFSET_STORAGE_TOPIC_CONFIG;
 import static 
org.apache.kafka.connect.runtime.distributed.DistributedConfig.STATUS_STORAGE_TOPIC_CONFIG;
 import static org.apache.kafka.connect.sink.SinkTask.TOPICS_CONFIG;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
@@ -896,6 +897,35 @@ public class WorkerTest {
         verifyKafkaClusterId();
     }
 
+    @ParameterizedTest
+    @ValueSource(booleans = {true, false})
+    public void testConnectorStatusMetricsGroup_tasksFailedToStart(boolean 
enableTopicCreation) {
+        setup(enableTopicCreation);
+        mockKafkaClusterId();
+        mockInternalConverters();
+        mockFileConfigProvider();
+
+        worker = new Worker(WORKER_ID,
+                new MockTime(),
+                plugins,
+                config,
+                offsetBackingStore,
+                noneConnectorClientConfigOverridePolicy);
+        worker.herder = herder;
+
+        // Pass an empty tasks map to simulate all tasks failing to start
+        Worker.ConnectorStatusMetricsGroup metricsGroup = new 
Worker.ConnectorStatusMetricsGroup(
+                worker.metrics(), new ConcurrentHashMap<>(), herder
+        );
+
+        ConnectorTaskId taskId1 = new ConnectorTaskId("c1", 0);
+        ConnectorTaskId taskId2 = new ConnectorTaskId("c1", 1);
+        metricsGroup.recordTaskAdded(taskId1);
+        metricsGroup.recordTaskAdded(taskId2);
+        metricsGroup.recordTaskRemoved(taskId1);
+        assertDoesNotThrow(() -> metricsGroup.recordTaskRemoved(taskId2), 
"should not throw NPE");
+    }
+
     @ParameterizedTest
     @ValueSource(booleans = {true, false})
     public void testStartTaskFailure(boolean enableTopicCreation) {