[fix] Unify fetch offset topic name (#1837)

The CLI needs to check the current committed topic’s name before
resetting. But in KoP it will transfer to a short topic name.

https://github.com/apache/kafka/blob/d46c3f259cce25c43f20fba3943d5cb34ed909ea/core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala#L818-L820

```
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092  --group kop/s1 --topic public/default/p3:0 --reset-offsets --shift-by -10 --execute
Cannot shift offset for partition public/default/p3-0 since there is no current committed offset
```
<img width="1120" alt="image"
src="https://user-images.githubusercontent.com/22697570/236761387-c06930c1-3072-43b2-84cb-7e16f2504999.png">

Return full name when list consumer group offsets.

(cherry picked from commit e58b7b9)

Author: Demogorgon314

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java

@@ -1109,16 +1109,14 @@ protected CompletableFuture<Void> storeGroupId(String groupId, String groupIdPat
     @VisibleForTesting
     public <T> void replaceTopicPartition(Map<TopicPartition, T> replacedMap,
                                           Map<TopicPartition, TopicPartition> replacingIndex) {
-        String namespacePrefix = currentNamespacePrefix();
         Map<TopicPartition, T> newMap = new HashMap<>();
         replacedMap.entrySet().removeIf(entry -> {
             if (replacingIndex.containsKey(entry.getKey())) {
                 newMap.put(replacingIndex.get(entry.getKey()), entry.getValue());
                 return true;
             } else if (KopTopic.isFullTopicName(entry.getKey().topic())) {
                 newMap.put(new TopicPartition(
-                                KopTopic.removeDefaultNamespacePrefix(entry.getKey().topic(),
-                                        namespacePrefix),
+                                KopTopic.removePersistentDomain(entry.getKey().topic()),
                                 entry.getKey().partition()),
                         entry.getValue());
                 return true;

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/KopTopic.java

@@ -42,6 +42,14 @@ public static String removeDefaultNamespacePrefix(String fullTopicName, String n
         }
     }
 
+    public static String removePersistentDomain(String fullTopicName) {
+        if (fullTopicName.startsWith(persistentDomain)) {
+            return fullTopicName.substring(persistentDomain.length());
+        } else {
+            return fullTopicName;
+        }
+    }
+
     @Getter
     private final String originalName;
     @Getter

tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandlerTest.java

@@ -833,7 +833,7 @@ public void testEmptyReplacingIndex() {
         final String topic = "test-topic";
 
         // 1. original tp
-        final TopicPartition tp0 = new TopicPartition(topic, 0);
+        final TopicPartition tp0 = new TopicPartition(namespace + "/" + topic, 0);
 
         // 2. full topic and tp
         final String fullNameTopic = "persistent://" + namespace + "/" + topic;
@@ -851,7 +851,7 @@ public void testEmptyReplacingIndex() {
         assertEquals(1, replacedMap.size());
 
         // 5. after replace, replacedMap has a short topic name
-        replacedMap.forEach(((topicPartition, s) -> assertEquals(tp0, topicPartition)));
+        replacedMap.forEach(((topicPartition, s) -> assertEquals(topicPartition, tp0)));
     }
 
     @Test
@@ -860,7 +860,7 @@ public void testNonEmptyReplacingIndex() {
         final String topic = "test-topic";
 
         // 1. original tp
-        final TopicPartition tp0 = new TopicPartition(topic, 0);
+        final TopicPartition tp0 = new TopicPartition(namespace + "/" + topic, 0);
 
         // 2. full topic and tp
         final String fullNameTopic = "persistent://" + namespace + "/" + topic;
@@ -879,7 +879,7 @@ public void testNonEmptyReplacingIndex() {
         assertEquals(1, replacedMap.size());
 
         // 5. after replace, replacedMap has a short topic name
-        replacedMap.forEach(((topicPartition, s) -> assertEquals(tp0, topicPartition)));
+        replacedMap.forEach(((topicPartition, s) -> assertEquals(topicPartition, tp0)));
     }
 
     @Test(timeOut = 20000000)
@@ -959,12 +959,10 @@ group2, new ListConsumerGroupOffsetsSpec().topicPartitions(null))
         //  topic name from offset fetch response must be short topic name
         offsetAndMetadataMapGroup1.keySet().forEach(topicPartition -> assertEquals(topic, topicPartition.topic()));
 
-
         consumer.close();
         consumer2.close();
         consumer2b.close();
         kafkaAdmin.close();
-
     }
 
     @Test(timeOut = 20000)

tests/src/test/java/io/streamnative/pulsar/handlers/kop/OffsetResetTest.java

@@ -26,6 +26,7 @@
 import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
@@ -79,7 +80,7 @@ protected void cleanup() throws Exception {
 
     @Test(timeOut = 30000, enabled = false)
     public void testGreaterThanEndOffset() throws Exception {
-        final String topic = "test-reset-offset-topic";
+        final String topic = "public/default/test-reset-offset-topic";
         final String group = "test-reset-offset-groupid";
         final int numPartitions = 1;
 
@@ -313,8 +314,9 @@ private long describeGroups(String group, String topic) {
                     .map(info -> new TopicPartition(info.topic(), info.partition())).collect(Collectors.toList())) {
                 log.info("offset part: {}", adminClient.listConsumerGroupOffsets(group)
                         .partitionsToOffsetAndMetadata().get());
-                long offset = adminClient.listConsumerGroupOffsets(group).partitionsToOffsetAndMetadata().get()
-                        .get(topicPartition).offset();
+                Map<TopicPartition, OffsetAndMetadata> topicPartitionOffsetAndMetadataMap =
+                        adminClient.listConsumerGroupOffsets(group).partitionsToOffsetAndMetadata().get();
+                long offset = topicPartitionOffsetAndMetadataMap.get(topicPartition).offset();
                 long leo = consumer.endOffsets(Collections.singletonList(topicPartition))
                         .get(topicPartition);
                 lag += (leo - offset);
@@ -386,7 +388,7 @@ private void readFromOffsetMessagePulsar() throws Exception {
 
     @Test(timeOut = 30000)
     public void testCliReset() throws Exception {
-        String topic = "test-reset-offset-topic";
+        String topic = "public/default/test-reset-offset-topic";
         final String group = "test-reset-offset-groupid";
         final int numPartitions = 10;
 
@@ -420,7 +422,7 @@ public void testCliReset() throws Exception {
         assertEquals(msgs, totalMsgs);
         kConsumer.getConsumer().commitSync();
         readFromOffsetMessagePulsar();
-        assertEquals(0, describeGroups(group, topic));
+        assertEquals(describeGroups(group, topic), 0);
 
         // simulate the consumer has closed
         kConsumer.close();