fix offset data loss

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

@@ -18,6 +18,7 @@
 import static io.streamnative.pulsar.handlers.kop.utils.TopicNameUtils.getKafkaTopicNameFromPulsarTopicname;
 import static org.apache.pulsar.common.naming.TopicName.PARTITIONED_TOPIC_SUFFIX;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableMap;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.socket.SocketChannel;
@@ -60,6 +61,7 @@
 import org.apache.pulsar.broker.service.BrokerService;
 import org.apache.pulsar.client.admin.Lookup;
 import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.common.naming.NamespaceBundle;
 import org.apache.pulsar.common.naming.NamespaceName;
 import org.apache.pulsar.common.naming.TopicName;
@@ -101,6 +103,11 @@ public GroupCoordinator getGroupCoordinator(String tenant) {
         return groupCoordinatorsByTenant.computeIfAbsent(tenant, this::createAndBootGroupCoordinator);
     }
 
+    @VisibleForTesting
+    public Map<String, GroupCoordinator> getGroupCoordinators() {
+        return groupCoordinatorsByTenant;
+    }
+
     @Override
     public TransactionCoordinator getTransactionCoordinator(String tenant) {
         return transactionCoordinatorByTenant.computeIfAbsent(tenant, this::createAndBootTransactionCoordinator);
@@ -485,11 +492,27 @@ private GroupCoordinator startGroupCoordinator(String tenant, SystemTopicClient
             kafkaConfig.getGroupInitialRebalanceDelayMs()
         );
 
+        String topicName = tenant + "/" + kafkaConfig.getKafkaMetadataNamespace()
+                + "/" + Topic.GROUP_METADATA_TOPIC_NAME;
+
+        PulsarAdmin pulsarAdmin;
+        int offsetTopicNumPartitions;
+        try {
+            pulsarAdmin = brokerService.getPulsar().getAdminClient();
+            offsetTopicNumPartitions = pulsarAdmin.topics().getPartitionedTopicMetadata(topicName).partitions;
+            if (offsetTopicNumPartitions == 0) {
+                log.error("Offset topic should not be a non-partitioned topic.");
+                throw new IllegalStateException("Offset topic should not be a non-partitioned topic.");
+            }
+        }  catch (PulsarServerException | PulsarAdminException e) {
+            log.error("Failed to get offset topic partition metadata .", e);
+            throw new IllegalStateException(e);
+        }
+
+
         OffsetConfig offsetConfig = OffsetConfig.builder()
-            .offsetsTopicName(tenant + "/"
-                + kafkaConfig.getKafkaMetadataNamespace()
-                + "/" + Topic.GROUP_METADATA_TOPIC_NAME)
-            .offsetsTopicNumPartitions(kafkaConfig.getOffsetsTopicNumPartitions())
+            .offsetsTopicName(topicName)
+            .offsetsTopicNumPartitions(offsetTopicNumPartitions)
             .offsetsTopicCompressionType(CompressionType.valueOf(kafkaConfig.getOffsetsTopicCompressionCodec()))
             .maxMetadataSize(kafkaConfig.getOffsetMetadataMaxSize())
             .offsetsRetentionCheckIntervalMs(kafkaConfig.getOffsetsRetentionCheckIntervalMs())

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/coordinator/group/GroupMetadataManager.java

@@ -26,6 +26,7 @@
 import static org.apache.kafka.common.internals.Topic.GROUP_METADATA_TOPIC_NAME;
 import static org.apache.pulsar.common.naming.TopicName.PARTITIONED_TOPIC_SUFFIX;
 
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
 import io.streamnative.pulsar.handlers.kop.coordinator.group.GroupMetadata.CommitRecordMetadataAndOffset;
@@ -1364,7 +1365,8 @@ boolean addLoadingPartition(int partition) {
      *
      * <p>Visible for testing
      */
-    boolean removeLoadingPartition(int partition) {
+    @VisibleForTesting
+    public boolean removeLoadingPartition(int partition) {
         return inLock(partitionLock, () -> loadingPartitions.remove(partition));
     }
 

tests/src/test/java/io/streamnative/pulsar/handlers/kop/coordinator/group/GroupMetadataManagerTest.java

@@ -286,6 +286,23 @@ private int completeTransactionalOffsetCommit(ByteBuffer buffer,
         return 1;
     }
 
+    @Test
+    public void testOffsetTopicNumPartitionsModify() {
+        int consumerGroupPartitionId =
+                GroupMetadataManager.getPartitionId(groupId, conf.getOffsetsTopicNumPartitions());
+        conf.setOffsetsTopicNumPartitions(100);
+
+        KafkaProtocolHandler handler = (KafkaProtocolHandler) pulsar.getProtocolHandlers().protocol("kafka");
+        // remove here to trigger a new creating for GroupCoordinator
+        handler.getGroupCoordinators().remove(conf.getKafkaMetadataTenant());
+        GroupMetadataManager newMetaManager =
+                handler.getGroupCoordinator(conf.getKafkaMetadataTenant()).getGroupManager();
+
+        int newPartitionsId =
+                GroupMetadataManager.getPartitionId(groupId, newMetaManager.offsetConfig().offsetsTopicNumPartitions());
+        assertEquals(consumerGroupPartitionId, newPartitionsId);
+    }
+
     @Test
     public void testLoadOffsetsWithoutGroup() throws Exception {
         Map<TopicPartition, Long> committedOffsets = new HashMap<>();
@@ -1032,17 +1049,17 @@ public void testOffsetWriteAfterGroupRemoved() throws Exception {
         ByteBuffer buffer = newMemoryRecordsBuffer(newOffsetCommitRecords);
 
         byte[] key = groupMetadataKey(groupId);
-
-        Producer<ByteBuffer> producer = groupMetadataManager.getOffsetsTopicProducer(groupPartitionId).get();
+        int consumerGroupPartitionId =
+                GroupMetadataManager.getPartitionId(groupId, conf.getOffsetsTopicNumPartitions());
+        Producer<ByteBuffer> producer = groupMetadataManager.getOffsetsTopicProducer(consumerGroupPartitionId).get();
         producer.newMessage()
             .keyBytes(key)
             .value(buffer)
             .eventTime(Time.SYSTEM.milliseconds())
             .send();
-
+        groupMetadataManager.removeLoadingPartition(consumerGroupPartitionId);
         CompletableFuture<GroupMetadata> onLoadedFuture = new CompletableFuture<>();
-        groupMetadataManager.scheduleLoadGroupAndOffsets(
-            groupPartitionId,
+        groupMetadataManager.scheduleLoadGroupAndOffsets(consumerGroupPartitionId,
             groupMetadata -> onLoadedFuture.complete(groupMetadata)
         ).get();
         GroupMetadata group = onLoadedFuture.get();