(kafka) branch trunk updated: KAFKA-19497; Topic replay code does not handle creation and deletion in the same delta (#20242)

[jsancio]

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

jsancio 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 92fe00e184d KAFKA-19497; Topic replay code does not handle creation 
and deletion in the same delta (#20242)
92fe00e184d is described below

commit 92fe00e184dbe4a32f01d8f283252ca3ec898d14
Author: anonymous <zt...@confluent.io>
AuthorDate: Wed Aug 27 15:34:12 2025 -0500

    KAFKA-19497; Topic replay code does not handle creation and deletion in the 
same delta (#20242)
    
    There is a small logic bug in topic replay. If a topic is created and
    then removed before the TopicsDelta is applied, we end up with the
    deleted topic in createdTopics on the delta. Tis issue is fixed by
    removing the topicName from createTopics when replaying
    RemoveTopicRecord to make sure the deleted topics won't appear in
    createTopics.
    
    Reviewers: José Armando García Sancio <jsan...@apache.org>, Kevin Wu
     <kevin.wu2...@gmail.com>, Alyssa Huang <ahu...@confluent.io>
---
 .../src/main/java/org/apache/kafka/image/TopicsDelta.java     |  1 +
 .../src/test/java/org/apache/kafka/image/TopicsImageTest.java | 11 +++++++++++
 2 files changed, 12 insertions(+)

diff --git a/metadata/src/main/java/org/apache/kafka/image/TopicsDelta.java 
b/metadata/src/main/java/org/apache/kafka/image/TopicsDelta.java
index 63ae2d06026..9b0ad7a916a 100644
--- a/metadata/src/main/java/org/apache/kafka/image/TopicsDelta.java
+++ b/metadata/src/main/java/org/apache/kafka/image/TopicsDelta.java
@@ -136,6 +136,7 @@ public final class TopicsDelta {
         String topicName;
         if (topicDelta != null) {
             topicName = topicDelta.image().name();
+            createdTopics.remove(topicName);
             if (image.topicsById().containsKey(record.topicId())) {
                 deletedTopicIds.add(record.topicId());
             }
diff --git a/metadata/src/test/java/org/apache/kafka/image/TopicsImageTest.java 
b/metadata/src/test/java/org/apache/kafka/image/TopicsImageTest.java
index 9154e799a33..bfdccc424e2 100644
--- a/metadata/src/test/java/org/apache/kafka/image/TopicsImageTest.java
+++ b/metadata/src/test/java/org/apache/kafka/image/TopicsImageTest.java
@@ -897,4 +897,15 @@ public class TopicsImageTest {
         var result = IMAGE1.topicIdToNameView().get("zar");
         assertNull(result);
     }
+
+    @Test
+    public void testTopicsDeltaCreateThenDelete() {
+        TopicsDelta delta = new TopicsDelta(TopicsImage.EMPTY);
+        delta.replay(new TopicRecord().setName("test").setTopicId(FOO_UUID));
+        assertEquals(delta.createdTopicIds().contains(FOO_UUID), true);
+        assertEquals(delta.deletedTopicIds().contains(FOO_UUID), false);
+        delta.replay(new RemoveTopicRecord().setTopicId(FOO_UUID));
+        assertEquals(delta.deletedTopicIds().contains(FOO_UUID), false);
+        assertEquals(delta.createdTopicIds().contains(FOO_UUID), false);
+    }
 }