support __consumer_offsets topic compaction (#401)

fix #316
streamnative · Mar 11, 2021 · 5ad37f3 · 5ad37f3
1 parent bc16dc1
commit 5ad37f3
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 0 deletions.
diff --git a/...main/java/io/streamnative/pulsar/handlers/kop/coordinator/group/GroupMetadataManager.java b/...main/java/io/streamnative/pulsar/handlers/kop/coordinator/group/GroupMetadataManager.java
@@ -1399,6 +1399,7 @@ CompletableFuture<Reader<ByteBuffer>> getOffsetsTopicReader(int partitionId) {
                 }
                 return metadataTopicReaderBuilder.clone()
                     .topic(offsetConfig.offsetsTopicName() + PARTITIONED_TOPIC_SUFFIX + partitionId)
+                    .readCompacted(true)
                     .createAsync();
             });
     }

diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/MetadataUtils.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/MetadataUtils.java
@@ -43,6 +43,8 @@
  */
 @Slf4j
 public class MetadataUtils {
+    // trigger compaction when the offset backlog reaches 100MB
+    public static final int MAX_COMPACTION_THRESHOLD = 100 * 1024 * 1024;
 
     public static String constructOffsetsTopicBaseName(KafkaServiceConfiguration conf) {
         return conf.getKafkaMetadataTenant() + "/" + conf.getKafkaMetadataNamespace()
@@ -112,6 +114,7 @@ public static void createKafkaMetadataIfMissing(PulsarAdmin pulsarAdmin, KafkaSe
                 namespaces.setNamespaceReplicationClusters(kafkaMetadataNamespace, replicationClusters);
                 namespaces.setRetention(kafkaMetadataNamespace,
                     new RetentionPolicies(-1, -1));
+                namespaces.setCompactionThreshold(kafkaMetadataNamespace, MAX_COMPACTION_THRESHOLD);
             } else {
                 List<String> replicationClusters = namespaces.getNamespaceReplicationClusters(kafkaMetadataNamespace);
                 if (!replicationClusters.contains(cluster)) {