(kafka) branch 3.9 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 3.9
in repository https://gitbox.apache.org/repos/asf/kafka.git


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

commit 2c0050baef66177d529d2f10dc525cc61a50d70a
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 0a44028a30d..22afbfafc77 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
@@ -2331,8 +2331,8 @@ public 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 bd7b980b408..0fb78e185fe 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
@@ -147,6 +147,7 @@ import static 
org.apache.kafka.connect.sink.SinkTask.TOPICS_CONFIG;
 import static org.hamcrest.CoreMatchers.containsString;
 import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.hamcrest.MatcherAssert.assertThat;
+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;
@@ -906,6 +907,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) {