KAFKA-19497; Topic replay code does not handle creation and deletion in the same delta (apache#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 <[email protected]>, Kevin Wu
 <[email protected]>, Alyssa Huang <[email protected]>

Author: 0xffff-zhiyan

metadata/src/main/java/org/apache/kafka/image/TopicsDelta.java

@@ -136,6 +136,7 @@ public String replay(RemoveTopicRecord record) {
         String topicName;
         if (topicDelta != null) {
             topicName = topicDelta.image().name();
+            createdTopics.remove(topicName);
             if (image.topicsById().containsKey(record.topicId())) {
                 deletedTopicIds.add(record.topicId());
             }

metadata/src/test/java/org/apache/kafka/image/TopicsImageTest.java

@@ -897,4 +897,15 @@ public void testTopicIdToNameView() {
         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);
+    }
 }