[FEATURE] Support kafka LogValidator validate inner records and compression codec when handle producer request with entryFormat=kafka

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

@@ -39,6 +39,7 @@
 import io.streamnative.pulsar.handlers.kop.exceptions.KoPTopicException;
 import io.streamnative.pulsar.handlers.kop.format.EntryFormatter;
 import io.streamnative.pulsar.handlers.kop.format.EntryFormatterFactory;
+import io.streamnative.pulsar.handlers.kop.format.KafkaEntryFormatter;
 import io.streamnative.pulsar.handlers.kop.offset.OffsetAndMetadata;
 import io.streamnative.pulsar.handlers.kop.offset.OffsetMetadata;
 import io.streamnative.pulsar.handlers.kop.security.SaslAuthenticator;
@@ -49,7 +50,9 @@
 import io.streamnative.pulsar.handlers.kop.security.auth.SimpleAclAuthorizer;
 import io.streamnative.pulsar.handlers.kop.stats.StatsLogger;
 import io.streamnative.pulsar.handlers.kop.utils.CoreUtils;
+import io.streamnative.pulsar.handlers.kop.utils.KopLogValidator;
 import io.streamnative.pulsar.handlers.kop.utils.KopTopic;
+import io.streamnative.pulsar.handlers.kop.utils.LongRef;
 import io.streamnative.pulsar.handlers.kop.utils.MessageIdUtils;
 import io.streamnative.pulsar.handlers.kop.utils.OffsetFinder;
 import io.streamnative.pulsar.handlers.kop.utils.ZooKeeperUtils;
@@ -76,6 +79,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
@@ -84,6 +88,7 @@
 import lombok.extern.slf4j.Slf4j;
 import org.apache.bookkeeper.common.util.MathUtils;
 import org.apache.bookkeeper.mledger.AsyncCallbacks;
+import org.apache.bookkeeper.mledger.ManagedLedger;
 import org.apache.bookkeeper.mledger.ManagedLedgerException;
 import org.apache.bookkeeper.mledger.Position;
 import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
@@ -97,12 +102,14 @@
 import org.apache.kafka.common.errors.LeaderNotAvailableException;
 import org.apache.kafka.common.protocol.ApiKeys;
 import org.apache.kafka.common.protocol.Errors;
+import org.apache.kafka.common.record.CompressionType;
 import org.apache.kafka.common.record.ControlRecordType;
 import org.apache.kafka.common.record.EndTransactionMarker;
 import org.apache.kafka.common.record.InvalidRecordException;
 import org.apache.kafka.common.record.MemoryRecords;
 import org.apache.kafka.common.record.MutableRecordBatch;
 import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.record.TimestampType;
 import org.apache.kafka.common.requests.AbstractRequest;
 import org.apache.kafka.common.requests.AbstractResponse;
 import org.apache.kafka.common.requests.AddOffsetsToTxnRequest;
@@ -233,6 +240,7 @@ public class KafkaRequestHandler extends KafkaCommandDecoder {
     private final long resumeThresholdPendingBytes;
     private final AtomicLong pendingBytes = new AtomicLong(0);
     private volatile boolean autoReadDisabledPublishBufferLimiting = false;
+    private final String brokerCompressionType;
 
     private String getCurrentTenant() {
         if (kafkaConfig.isKafkaEnableMultiTenantMetadata()
@@ -312,6 +320,7 @@ public KafkaRequestHandler(PulsarService pulsarService,
         this.maxPendingBytes = kafkaConfig.getMaxMessagePublishBufferSizeInMB() * 1024L * 1024L;
         this.resumeThresholdPendingBytes = this.maxPendingBytes / 2;
         this.failedAuthenticationDelayMs = kafkaConfig.getFailedAuthenticationDelayMs();
+        this.brokerCompressionType = kafkaConfig.getKafkaCompressionType();
 
         // update alive channel count stats
         RequestStats.ALIVE_CHANNEL_COUNT_INSTANCE.incrementAndGet();
@@ -1007,17 +1016,6 @@ private void handlePartitionRecords(final KafkaHeaderAndRequest produceHar,
             final long beforeRecordsProcess = MathUtils.nowInNano();
             final MemoryRecords validRecords =
                     validateRecords(produceHar.getHeader().apiVersion(), topicPartition, records);
-            final int numMessages = EntryFormatter.parseNumMessages(validRecords);
-            final ByteBuf byteBuf = entryFormatter.encode(validRecords, numMessages);
-            requestStats.getProduceEncodeStats().registerSuccessfulEvent(
-                    MathUtils.elapsedNanos(beforeRecordsProcess), TimeUnit.NANOSECONDS);
-            startSendOperationForThrottling(byteBuf.readableBytes());
-
-            if (log.isDebugEnabled()) {
-                log.debug("[{}] Request {}: Produce messages for topic {} partition {}, "
-                                + "request size: {} ", ctx.channel(), produceHar.getHeader(),
-                        topicPartition.topic(), topicPartition.partition(), numPartitions);
-            }
 
             final CompletableFuture<Optional<PersistentTopic>> topicFuture =
                     topicManager.getTopic(fullPartitionName);
@@ -1034,7 +1032,46 @@ private void handlePartitionRecords(final KafkaHeaderAndRequest produceHar,
             }
 
             final Consumer<Optional<PersistentTopic>> persistentTopicConsumer = persistentTopicOpt -> {
-                publishMessages(persistentTopicOpt, byteBuf, numMessages, validRecords, topicPartition,
+                if (!persistentTopicOpt.isPresent()) {
+                    errorsConsumer.accept(Errors.NOT_LEADER_FOR_PARTITION);
+                    return;
+                }
+
+                final MemoryRecords finalValidRecords;
+                if (entryFormatter instanceof KafkaEntryFormatter) {
+                    final KopLogValidator.CompressionCodec sourceCodec = getSourceCodec(validRecords);
+                    final KopLogValidator.CompressionCodec targetCodec = getTargetCodec(sourceCodec);
+
+                    ManagedLedger managedLedger = persistentTopicOpt.get().getManagedLedger();
+                    long logEndOffset = MessageIdUtils.getLogEndOffset(managedLedger);
+                    LongRef offset = new LongRef(logEndOffset);
+
+                    finalValidRecords = KopLogValidator.validateMessagesAndAssignOffsets(validRecords,
+                            offset,
+                            System.currentTimeMillis(),
+                            sourceCodec,
+                            targetCodec,
+                            false,
+                            RecordBatch.MAGIC_VALUE_V2,
+                            TimestampType.CREATE_TIME,
+                            Long.MAX_VALUE);
+                } else {
+                    finalValidRecords = validRecords;
+                }
+
+                final int numMessages = EntryFormatter.parseNumMessages(finalValidRecords);
+                final ByteBuf byteBuf = entryFormatter.encode(finalValidRecords, numMessages);
+                requestStats.getProduceEncodeStats().registerSuccessfulEvent(
+                        MathUtils.elapsedNanos(beforeRecordsProcess), TimeUnit.NANOSECONDS);
+                startSendOperationForThrottling(byteBuf.readableBytes());
+
+                if (log.isDebugEnabled()) {
+                    log.debug("[{}] Request {}: Produce messages for topic {} partition {}, "
+                                    + "request size: {} ", ctx.channel(), produceHar.getHeader(),
+                            topicPartition.topic(), topicPartition.partition(), numPartitions);
+                }
+
+                publishMessages(persistentTopicOpt, byteBuf, numMessages, finalValidRecords, topicPartition,
                         offsetConsumer, errorsConsumer);
             };
 
@@ -2534,6 +2571,31 @@ private static MemoryRecords validateRecords(short version, TopicPartition topic
         return validRecords;
     }
 
+    private KopLogValidator.CompressionCodec getSourceCodec(MemoryRecords records) {
+        AtomicReference<KopLogValidator.CompressionCodec> sourceCodec =
+                new AtomicReference<>(new KopLogValidator.CompressionCodec(
+                        CompressionType.NONE.name, CompressionType.NONE.id));
+        records.batches().forEach(batch -> {
+            CompressionType compressionType = CompressionType.forId(batch.compressionType().id);
+            KopLogValidator.CompressionCodec messageCodec = new KopLogValidator.CompressionCodec(
+                    compressionType.name, compressionType.id);
+            if (!messageCodec.name().equals("none")) {
+                sourceCodec.set(messageCodec);
+            }
+        });
+
+        return sourceCodec.get();
+    }
+
+    private KopLogValidator.CompressionCodec getTargetCodec(KopLogValidator.CompressionCodec sourceCodec) {
+        if (brokerCompressionType.equals("producer")) {
+            return sourceCodec;
+        } else {
+            CompressionType compressionType = CompressionType.forName(brokerCompressionType);
+            return new KopLogValidator.CompressionCodec(compressionType.name, compressionType.id);
+        }
+    }
+
     private void updateProducerStats(final TopicPartition topicPartition, final int numMessages, final int numBytes) {
         requestStats.getStatsLogger()
                 .scopeLabel(TOPIC_SCOPE, topicPartition.topic())

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

@@ -382,6 +382,13 @@ public class KafkaServiceConfiguration extends ServiceConfiguration {
     )
     private Set<String> kopAllowedNamespaces;
 
+    @FieldContext(
+            category = CATEGORY_KOP,
+            doc = "KOP server compression type. Only used for entryFormat=kafka. If it's not set to producer. "
+                    + "The client messages will be used compression type which configured in here."
+    )
+    private String kafkaCompressionType = "producer";
+
     private String checkAdvertisedListeners(String advertisedListeners) {
         StringBuilder listenersReBuilder = new StringBuilder();
         for (String listener : advertisedListeners.split(EndPoint.END_POINT_SEPARATOR)) {

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

@@ -21,7 +21,6 @@
 import io.netty.util.ReferenceCounted;
 import io.streamnative.pulsar.handlers.kop.exceptions.KoPMessageMetadataNotFoundException;
 import io.streamnative.pulsar.handlers.kop.utils.ByteBufUtils;
-import io.streamnative.pulsar.handlers.kop.utils.KopRecordsUtil;
 import io.streamnative.pulsar.handlers.kop.utils.MessageIdUtils;
 import java.io.IOException;
 import java.util.ArrayList;
@@ -32,7 +31,7 @@
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.record.ConvertedRecords;
 import org.apache.kafka.common.record.MemoryRecords;
-import org.apache.kafka.common.record.RecordBatch;
+import org.apache.kafka.common.utils.Time;
 import org.apache.pulsar.common.allocator.PulsarByteBufAllocator;
 import org.apache.pulsar.common.api.proto.KeyValue;
 import org.apache.pulsar.common.api.proto.MessageMetadata;
@@ -47,6 +46,7 @@ public class KafkaEntryFormatter implements EntryFormatter {
     // These key-value identifies the entry's format as kafka
     private static final String IDENTITY_KEY = "entry.format";
     private static final String IDENTITY_VALUE = EntryFormatterFactory.EntryFormat.KAFKA.name().toLowerCase();
+    private final Time time = Time.SYSTEM;
 
     @Override
     public ByteBuf encode(MemoryRecords records, int numMessages) {
@@ -77,11 +77,11 @@ public DecodeResult decode(List<Entry> entries, byte magic) {
 
                     // batch magic greater than the magic corresponding to the version requested by the client
                     // need down converted
-                    if (batchMagic > magic || batchMagic != RecordBatch.MAGIC_VALUE_V2) {
+                    if (batchMagic > magic) {
                         MemoryRecords memoryRecords = MemoryRecords.readableRecords(ByteBufUtils.getNioBuffer(byteBuf));
                         //down converted, batch magic will be set to client magic
                         ConvertedRecords<MemoryRecords> convertedRecords =
-                                KopRecordsUtil.convertAndAssignOffsets(memoryRecords.batches(), magic, startOffset);
+                                memoryRecords.downConvert(magic, startOffset, time);
 
                         final ByteBuf kafkaBuffer = Unpooled.wrappedBuffer(convertedRecords.records().buffer());
                         orderedByteBuf.add(kafkaBuffer);
@@ -91,8 +91,7 @@ public DecodeResult decode(List<Entry> entries, byte magic) {
                         optionalByteBufs.ifPresent(byteBufs -> byteBufs.add(kafkaBuffer));
 
                         if (log.isTraceEnabled()) {
-                            log.trace("[{}:{}] convertAndAssignOffsets record for down converted"
-                                            + " or assign offsets with v0 and v1 magic, start offset {},"
+                            log.trace("[{}:{}] MemoryRecords down converted, start offset {},"
                                             + " entry magic: {}, client magic: {}",
                                     entry.getLedgerId(), entry.getEntryId(), startOffset, batchMagic, magic);
                         }