[kafka] branch trunk updated: KAFKA-12525: Ignoring stale status statuses when reading from Connect status topic (#13453)

[kk...@apache.org]

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

kkarantasis 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 adacfea2d6c KAFKA-12525: Ignoring stale status statuses when reading from Connect status topic (#13453)
adacfea2d6c is described below

commit adacfea2d6c133c9bb2a7ac3891dc69eb6f91e03
Author: vamossagar12 <sa...@gmail.com>
AuthorDate: Tue Jul 11 20:35:10 2023 +0530

    KAFKA-12525: Ignoring stale status statuses when reading from Connect status topic (#13453)
    
    During fast consecutive rebalances where a task is revoked from one worker and assigned to another one, it has been observed that there is a small time window and thus a race condition during which a RUNNING status record in the new generation is produced and is immediately followed by a delayed UNASSIGNED status record belonging to the same or a previous generation before the worker that sends this message reads the RUNNING status record that corresponds to the latest generation.
    Although this doesn't inhibit the actual execution of tasks, it reports an incorrect status for those tasks(i.e UNASSIGNED). If the users have setup some kind of monitoring on tasks status then this could lead to false alarms for example.
    This fix addresses this problem by checking if a status message is stale after reading it and updates it's status only when it is safe to.
    
    Reviewers: Lucent-Wong <ma...@live.cn>, Chris Egerton <ch...@aiven.io>, Yash Mayya <ya...@gmail.com>, Konstantine Karantasis <k....@gmail.com>
---
 .../connect/storage/KafkaStatusBackingStore.java   | 19 ++++++++
 .../storage/KafkaStatusBackingStoreTest.java       | 52 ++++++++++++++++++++++
 2 files changed, 71 insertions(+)

diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaStatusBackingStore.java b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaStatusBackingStore.java
index 20f47db2262..d1580e3a7c1 100644
--- a/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaStatusBackingStore.java
+++ b/connect/runtime/src/main/java/org/apache/kafka/connect/storage/KafkaStatusBackingStore.java
@@ -588,6 +588,25 @@ public class KafkaStatusBackingStore extends KafkaTopicBasedBackingStore impleme
         synchronized (this) {
             log.trace("Received task {} status update {}", id, status);
             CacheEntry<TaskStatus> entry = getOrAdd(id);
+
+            // During frequent rebalances, there could be a race condition because of which
+            // an UNASSIGNED state of a prior or same generation can be sent by a worker despite a
+            // RUNNING status by another worker because the first worker
+            // couldn't read the latest RUNNING status. This can lead to an inaccurate status
+            // representation even though the task might be actually running.
+            // Note that this could also mean that when a generation reset happens, and an
+            // UNASSIGNED status is sent, then it would be ignored if the current status is RUNNING
+            // at a higher generation. But since it will be followed by a RUNNING or a different
+            // status message(at the lower generation) soon after, the misrepresentation of the UNASSIGNED
+            // status would be short-lived in most cases.
+            if (status.state() == TaskStatus.State.UNASSIGNED
+                    && entry.get() != null
+                    && entry.get().state() == TaskStatus.State.RUNNING
+                    && entry.get().generation() >= status.generation()) {
+                log.trace("Ignoring stale status {} in favour of more upto date status {}", status, entry.get());
+                return;
+            }
+
             entry.put(status);
         }
     }
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaStatusBackingStoreTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaStatusBackingStoreTest.java
index 71fe342ee28..4cad728ae65 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaStatusBackingStoreTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/storage/KafkaStatusBackingStoreTest.java
@@ -309,6 +309,58 @@ public class KafkaStatusBackingStoreTest {
         assertEquals(status, store.get(TASK));
     }
 
+    @Test
+    public void readTaskStateShouldIgnoreStaleStatusesFromOtherWorkers() {
+        byte[] value = new byte[0];
+        String otherWorkerId = "anotherhost:8083";
+        String yetAnotherWorkerId = "yetanotherhost:8083";
+
+        // This worker sends a RUNNING status in the most recent generation
+        Map<String, Object> firstStatusRead = new HashMap<>();
+        firstStatusRead.put("worker_id", otherWorkerId);
+        firstStatusRead.put("state", "RUNNING");
+        firstStatusRead.put("generation", 10L);
+
+        // Another worker still ends up producing an UNASSIGNED status before it could
+        // read the newer RUNNING status from above belonging to an older generation.
+        Map<String, Object> secondStatusRead = new HashMap<>();
+        secondStatusRead.put("worker_id", WORKER_ID);
+        secondStatusRead.put("state", "UNASSIGNED");
+        secondStatusRead.put("generation", 9L);
+
+        Map<String, Object> thirdStatusRead = new HashMap<>();
+        thirdStatusRead.put("worker_id", yetAnotherWorkerId);
+        thirdStatusRead.put("state", "RUNNING");
+        thirdStatusRead.put("generation", 1L);
+
+        when(converter.toConnectData(STATUS_TOPIC, value))
+                .thenReturn(new SchemaAndValue(null, firstStatusRead))
+                .thenReturn(new SchemaAndValue(null, secondStatusRead))
+                .thenReturn(new SchemaAndValue(null, thirdStatusRead));
+
+        when(converter.fromConnectData(eq(STATUS_TOPIC), any(Schema.class), any(Struct.class)))
+                .thenReturn(value);
+
+        doAnswer(invocation -> {
+            ((Callback) invocation.getArgument(2)).onCompletion(null, null);
+            store.read(consumerRecord(2, "status-task-conn-0", value));
+            return null;
+        }).when(kafkaBasedLog).send(eq("status-task-conn-0"), eq(value), any(Callback.class));
+
+        store.read(consumerRecord(0, "status-task-conn-0", value));
+        store.read(consumerRecord(1, "status-task-conn-0", value));
+
+        // The latest task status should reflect RUNNING status from the newer generation
+        TaskStatus status = new TaskStatus(TASK, TaskStatus.State.RUNNING, otherWorkerId, 10);
+        assertEquals(status, store.get(TASK));
+
+        // This  status is from the another worker not necessarily belonging to the above group.
+        // In this case, the status should get updated irrespective of whatever status info was present before.
+        TaskStatus latestStatus = new TaskStatus(TASK, TaskStatus.State.RUNNING, yetAnotherWorkerId, 1);
+        store.put(latestStatus);
+        assertEquals(latestStatus, store.get(TASK));
+    }
+
     @Test
     public void deleteConnectorState() {
         final byte[] value = new byte[0];