diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/DelayedProduceAndFetch.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/DelayedProduceAndFetch.java index a1070e0377..be68b7b4b2 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/DelayedProduceAndFetch.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/DelayedProduceAndFetch.java @@ -20,12 +20,12 @@ /** * A delayed create topic operation that is stored in the topic purgatory. */ -class DelayedProduceAndFetch extends DelayedOperation { +public class DelayedProduceAndFetch extends DelayedOperation { private final AtomicInteger topicPartitionNum; private final Runnable callback; - DelayedProduceAndFetch(long delayMs, AtomicInteger topicPartitionNum, Runnable callback) { + public DelayedProduceAndFetch(long delayMs, AtomicInteger topicPartitionNum, Runnable callback) { super(delayMs, Optional.empty()); this.topicPartitionNum = topicPartitionNum; this.callback = callback; diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaProtocolHandler.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaProtocolHandler.java index 73463ea946..2382ad66df 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaProtocolHandler.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaProtocolHandler.java @@ -26,8 +26,11 @@ import io.streamnative.pulsar.handlers.kop.coordinator.group.OffsetConfig; import io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionConfig; import io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionCoordinator; +import io.streamnative.pulsar.handlers.kop.format.EntryFormatter; +import io.streamnative.pulsar.handlers.kop.format.EntryFormatterFactory; import io.streamnative.pulsar.handlers.kop.stats.PrometheusMetricsProvider; import io.streamnative.pulsar.handlers.kop.stats.StatsLogger; +import io.streamnative.pulsar.handlers.kop.storage.ReplicaManager; import io.streamnative.pulsar.handlers.kop.utils.ConfigurationUtils; import io.streamnative.pulsar.handlers.kop.utils.MetadataUtils; import io.streamnative.pulsar.handlers.kop.utils.delayed.DelayedOperation; @@ -97,6 +100,7 @@ public class KafkaProtocolHandler implements ProtocolHandler, TenantContextManag private final Map groupCoordinatorsByTenant = new ConcurrentHashMap<>(); private final Map transactionCoordinatorByTenant = new ConcurrentHashMap<>(); + private final Map replicaManagerByTenant = new ConcurrentHashMap<>(); @Override public GroupCoordinator getGroupCoordinator(String tenant) { @@ -113,6 +117,29 @@ public TransactionCoordinator getTransactionCoordinator(String tenant) { return transactionCoordinatorByTenant.computeIfAbsent(tenant, this::createAndBootTransactionCoordinator); } + @Override + public ReplicaManager getReplicaManager(String tenant) { + return replicaManagerByTenant.computeIfAbsent(tenant, s -> { + Optional transactionCoordinatorOptional = Optional.empty(); + if (kafkaConfig.isEnableTransactionCoordinator()) { + transactionCoordinatorOptional = Optional.of(getTransactionCoordinator(tenant)); + } + EntryFormatter entryFormatter; + try { + entryFormatter = EntryFormatterFactory.create(kafkaConfig); + } catch (IllegalArgumentException e) { + log.error("Failed to init create enter formatter {}", tenant, e); + throw new IllegalStateException(e); + } + return new ReplicaManager( + kafkaConfig, + Time.SYSTEM, + entryFormatter, + transactionCoordinatorOptional, + producePurgatory); + }); + } + /** * Listener for invalidating the global Broker ownership cache. */ diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java index d411550786..c0e2a25fa4 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java @@ -19,7 +19,6 @@ import static io.streamnative.pulsar.handlers.kop.KafkaServiceConfiguration.TENANT_PLACEHOLDER; import static java.nio.charset.StandardCharsets.UTF_8; import static org.apache.kafka.common.internals.Topic.GROUP_METADATA_TOPIC_NAME; -import static org.apache.kafka.common.internals.Topic.TRANSACTION_STATE_TOPIC_NAME; import static org.apache.kafka.common.protocol.CommonFields.THROTTLE_TIME_MS; import static org.apache.kafka.common.requests.CreateTopicsRequest.TopicDetails; @@ -35,11 +34,8 @@ import io.streamnative.pulsar.handlers.kop.coordinator.transaction.AbortedIndexEntry; import io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionCoordinator; import io.streamnative.pulsar.handlers.kop.exceptions.KoPTopicException; -import io.streamnative.pulsar.handlers.kop.format.EncodeRequest; -import io.streamnative.pulsar.handlers.kop.format.EncodeResult; import io.streamnative.pulsar.handlers.kop.format.EntryFormatter; import io.streamnative.pulsar.handlers.kop.format.EntryFormatterFactory; -import io.streamnative.pulsar.handlers.kop.format.KafkaMixedEntryFormatter; 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,6 +45,8 @@ import io.streamnative.pulsar.handlers.kop.security.auth.ResourceType; import io.streamnative.pulsar.handlers.kop.security.auth.SimpleAclAuthorizer; import io.streamnative.pulsar.handlers.kop.stats.StatsLogger; +import io.streamnative.pulsar.handlers.kop.storage.AppendRecordsContext; +import io.streamnative.pulsar.handlers.kop.storage.ReplicaManager; import io.streamnative.pulsar.handlers.kop.utils.CoreUtils; import io.streamnative.pulsar.handlers.kop.utils.GroupIdUtils; import io.streamnative.pulsar.handlers.kop.utils.KafkaRequestUtils; @@ -58,7 +56,6 @@ import io.streamnative.pulsar.handlers.kop.utils.OffsetFinder; import io.streamnative.pulsar.handlers.kop.utils.TopicNameUtils; import io.streamnative.pulsar.handlers.kop.utils.delayed.DelayedOperation; -import io.streamnative.pulsar.handlers.kop.utils.delayed.DelayedOperationKey; import io.streamnative.pulsar.handlers.kop.utils.delayed.DelayedOperationPurgatory; import java.net.InetSocketAddress; import java.nio.ByteBuffer; @@ -67,7 +64,6 @@ import java.util.Collections; import java.util.HashMap; import java.util.HashSet; -import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Optional; @@ -86,9 +82,7 @@ import java.util.stream.IntStream; import lombok.Getter; 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; @@ -101,16 +95,12 @@ import org.apache.kafka.common.acl.AclOperation; import org.apache.kafka.common.config.ConfigResource; import org.apache.kafka.common.errors.AuthenticationException; -import org.apache.kafka.common.errors.CorruptRecordException; import org.apache.kafka.common.errors.LeaderNotAvailableException; -import org.apache.kafka.common.errors.RecordTooLargeException; import org.apache.kafka.common.protocol.ApiKeys; import org.apache.kafka.common.protocol.Errors; 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.requests.AbstractRequest; import org.apache.kafka.common.requests.AbstractResponse; @@ -279,6 +269,10 @@ public TransactionCoordinator getTransactionCoordinator() { return tenantContextManager.getTransactionCoordinator(getCurrentTenant()); } + public ReplicaManager getReplicaManager() { + return tenantContextManager.getReplicaManager(getCurrentTenant()); + } + public KafkaRequestHandler(PulsarService pulsarService, KafkaServiceConfiguration kafkaConfig, TenantContextManager tenantContextManager, @@ -472,12 +466,6 @@ private CompletableFuture getPartitionedTopicMetadataA return admin.topics().getPartitionedTopicMetadataAsync(topicName); } - private static boolean isInternalTopic(final String fullTopicName) { - String partitionedTopicName = TopicName.get(fullTopicName).getPartitionedTopicName(); - return partitionedTopicName.endsWith("/" + GROUP_METADATA_TOPIC_NAME) - || partitionedTopicName.endsWith("/" + TRANSACTION_STATE_TOPIC_NAME); - } - private CompletableFuture> expandAllowedNamespaces(Set allowedNamespaces) { String currentTenant = getCurrentTenant(kafkaConfig.getKafkaTenant()); return expandAllowedNamespaces(allowedNamespaces, currentTenant, pulsarService); @@ -599,7 +587,7 @@ protected void handleTopicMetadataRequest(KafkaHeaderAndRequest metadataHar, allTopicMetadata.add(new TopicMetadata( Errors.TOPIC_AUTHORIZATION_FAILED, topic, - isInternalTopic(topicName.toString()), + KopTopic.isInternalTopic(topicName.toString()), Collections.emptyList())); return; } @@ -644,7 +632,7 @@ protected void handleTopicMetadataRequest(KafkaHeaderAndRequest metadataHar, allTopicMetadata.add(new TopicMetadata( Errors.TOPIC_AUTHORIZATION_FAILED, topic, - isInternalTopic(fullTopicName), + KopTopic.isInternalTopic(fullTopicName), Collections.emptyList())); completeOneTopic.run(); }; @@ -699,7 +687,7 @@ protected void handleTopicMetadataRequest(KafkaHeaderAndRequest metadataHar, new TopicMetadata( Errors.UNKNOWN_TOPIC_OR_PARTITION, topic, - isInternalTopic(fullTopicName), + KopTopic.isInternalTopic(fullTopicName), Collections.emptyList())); completeOneTopic.run(); } @@ -709,7 +697,7 @@ protected void handleTopicMetadataRequest(KafkaHeaderAndRequest metadataHar, new TopicMetadata( Errors.UNKNOWN_TOPIC_OR_PARTITION, topic, - isInternalTopic(fullTopicName), + KopTopic.isInternalTopic(fullTopicName), Collections.emptyList())); log.warn("[{}] Request {}: Failed to get partitioned pulsar topic {} " + "metadata: {}", @@ -814,7 +802,8 @@ protected void handleTopicMetadataRequest(KafkaHeaderAndRequest metadataHar, // The topic returned to Kafka clients should be // the same with what it sent topic, - isInternalTopic(new KopTopic(topic, namespacePrefix).getFullName()), + KopTopic.isInternalTopic( + new KopTopic(topic, namespacePrefix).getFullName()), partitionMetadatas)); // whether completed all the topics requests. @@ -894,60 +883,6 @@ private void completeSendOperationForThrottling(long msgSize) { } } - private void publishMessages(final Optional persistentTopicOpt, - final EncodeResult encodeResult, - final TopicPartition topicPartition, - final Consumer offsetConsumer, - final Consumer errorsConsumer) { - final MemoryRecords records = encodeResult.getRecords(); - final int numMessages = encodeResult.getNumMessages(); - final ByteBuf byteBuf = encodeResult.getEncodedByteBuf(); - if (!persistentTopicOpt.isPresent()) { - encodeResult.recycle(); - // It will trigger a retry send of Kafka client - errorsConsumer.accept(Errors.NOT_LEADER_FOR_PARTITION); - return; - } - PersistentTopic persistentTopic = persistentTopicOpt.get(); - if (persistentTopic.isSystemTopic()) { - encodeResult.recycle(); - log.error("Not support producing message to system topic: {}", persistentTopic); - errorsConsumer.accept(Errors.INVALID_TOPIC_EXCEPTION); - return; - } - String namespacePrefix = currentNamespacePrefix(); - final String partitionName = KopTopic.toString(topicPartition, namespacePrefix); - - topicManager.registerProducerInPersistentTopic(partitionName, persistentTopic); - // collect metrics - encodeResult.updateProducerStats(topicPartition, requestStats, namespacePrefix); - - // publish - final CompletableFuture offsetFuture = new CompletableFuture<>(); - final long beforePublish = MathUtils.nowInNano(); - persistentTopic.publishMessage(byteBuf, - MessagePublishContext.get(offsetFuture, persistentTopic, numMessages, System.nanoTime())); - final RecordBatch batch = records.batchIterator().next(); - offsetFuture.whenComplete((offset, e) -> { - completeSendOperationForThrottling(byteBuf.readableBytes()); - encodeResult.recycle(); - if (e == null) { - if (batch.isTransactional()) { - getTransactionCoordinator().addActivePidOffset(TopicName.get(partitionName), batch.producerId(), - offset); - } - requestStats.getMessagePublishStats().registerSuccessfulEvent( - MathUtils.elapsedNanos(beforePublish), TimeUnit.NANOSECONDS); - offsetConsumer.accept(offset); - } else { - log.error("publishMessages for topic partition: {} failed when write.", partitionName, e); - requestStats.getMessagePublishStats().registerFailedEvent( - MathUtils.elapsedNanos(beforePublish), TimeUnit.NANOSECONDS); - errorsConsumer.accept(Errors.KAFKA_STORAGE_ERROR); - } - }); - } - @Override protected void handleProduceRequest(KafkaHeaderAndRequest produceHar, CompletableFuture resultFuture) { @@ -955,170 +890,73 @@ protected void handleProduceRequest(KafkaHeaderAndRequest produceHar, ProduceRequest produceRequest = (ProduceRequest) produceHar.getRequest(); final int numPartitions = produceRequest.partitionRecordsOrFail().size(); - - final Map responseMap = new ConcurrentHashMap<>(); - // delay produce - final AtomicInteger topicPartitionNum = new AtomicInteger(produceRequest.partitionRecordsOrFail().size()); + if (numPartitions == 0) { + resultFuture.complete(new ProduceResponse(Collections.emptyMap())); + return; + } + final Map unauthorizedTopicResponsesMap = new ConcurrentHashMap<>(); + final Map authorizedRequestInfo = new ConcurrentHashMap<>(); int timeoutMs = produceRequest.timeout(); - Runnable complete = () -> { - topicPartitionNum.set(0); - if (resultFuture.isDone()) { - // It may be triggered again in DelayedProduceAndFetch - return; - } - // add the topicPartition with timeout error if it's not existed in responseMap - produceRequest.partitionRecordsOrFail().keySet().forEach(topicPartition -> { - if (!responseMap.containsKey(topicPartition)) { - responseMap.put(topicPartition, new PartitionResponse(Errors.REQUEST_TIMED_OUT)); + String namespacePrefix = currentNamespacePrefix(); + final AtomicInteger unfinishedAuthorizationCount = new AtomicInteger(numPartitions); + Runnable completeOne = () -> { + // When complete one authorization or failed, will do the action first. + if (unfinishedAuthorizationCount.decrementAndGet() == 0) { + if (authorizedRequestInfo.isEmpty()) { + resultFuture.complete(new ProduceResponse(unauthorizedTopicResponsesMap)); + return; } - }); - if (log.isDebugEnabled()) { - log.debug("[{}] Request {}: Complete handle produce.", ctx.channel(), produceHar.toString()); - } - resultFuture.complete(new ProduceResponse(responseMap)); - }; - BiConsumer addPartitionResponse = (topicPartition, response) -> { - responseMap.put(topicPartition, response); - // reset topicPartitionNum - int restTopicPartitionNum = topicPartitionNum.decrementAndGet(); - if (restTopicPartitionNum < 0) { - return; - } - if (restTopicPartitionNum == 0) { - complete.run(); + AppendRecordsContext appendRecordsContext = AppendRecordsContext.get( + topicManager, + requestStats, + this::startSendOperationForThrottling, + this::completeSendOperationForThrottling, + pendingTopicFuturesMap); + getReplicaManager().appendRecords( + timeoutMs, + false, + produceHar.getRequest().version(), + namespacePrefix, + authorizedRequestInfo, + appendRecordsContext + ).whenComplete((response, ex) -> { + appendRecordsContext.recycle(); + if (ex != null) { + resultFuture.completeExceptionally(ex.getCause()); + return; + } + Map mergedResponse = new HashMap<>(); + mergedResponse.putAll(response); + mergedResponse.putAll(unauthorizedTopicResponsesMap); + resultFuture.complete(new ProduceResponse(mergedResponse)); + }); } }; - String namespacePrefix = currentNamespacePrefix(); produceRequest.partitionRecordsOrFail().forEach((topicPartition, records) -> { - final Consumer offsetConsumer = offset -> addPartitionResponse.accept( - topicPartition, new PartitionResponse(Errors.NONE, offset, -1L, -1L)); - final Consumer errorsConsumer = - errors -> addPartitionResponse.accept(topicPartition, new PartitionResponse(errors)); - final Consumer exceptionConsumer = - e -> addPartitionResponse.accept(topicPartition, new PartitionResponse(Errors.forException(e))); final String fullPartitionName = KopTopic.toString(topicPartition, namespacePrefix); - authorize(AclOperation.WRITE, Resource.of(ResourceType.TOPIC, fullPartitionName)) .whenComplete((isAuthorized, ex) -> { if (ex != null) { log.error("Write topic authorize failed, topic - {}. {}", fullPartitionName, ex.getMessage()); - errorsConsumer.accept(Errors.TOPIC_AUTHORIZATION_FAILED); + unauthorizedTopicResponsesMap.put(topicPartition, + new ProduceResponse.PartitionResponse(Errors.TOPIC_AUTHORIZATION_FAILED)); + completeOne.run(); return; } if (!isAuthorized) { - errorsConsumer.accept(Errors.TOPIC_AUTHORIZATION_FAILED); + unauthorizedTopicResponsesMap.put(topicPartition, + new ProduceResponse.PartitionResponse(Errors.TOPIC_AUTHORIZATION_FAILED)); + completeOne.run(); return; } - - handlePartitionRecords(produceHar, - topicPartition, - records, - numPartitions, - fullPartitionName, - offsetConsumer, - errorsConsumer, - exceptionConsumer); + authorizedRequestInfo.put(topicPartition, records); + completeOne.run(); }); }); - // delay produce - if (timeoutMs <= 0) { - complete.run(); - } else { - List delayedCreateKeys = - produceRequest.partitionRecordsOrFail().keySet().stream() - .map(DelayedOperationKey.TopicPartitionOperationKey::new).collect(Collectors.toList()); - DelayedProduceAndFetch delayedProduce = new DelayedProduceAndFetch(timeoutMs, topicPartitionNum, complete); - producePurgatory.tryCompleteElseWatch(delayedProduce, delayedCreateKeys); - } - } - private void handlePartitionRecords(final KafkaHeaderAndRequest produceHar, - final TopicPartition topicPartition, - final MemoryRecords records, - final int numPartitions, - final String fullPartitionName, - final Consumer offsetConsumer, - final Consumer errorsConsumer, - final Consumer exceptionConsumer) { - // check KOP inner topic - if (isInternalTopic(fullPartitionName)) { - log.error("[{}] Request {}: not support produce message to inner topic. topic: {}", - ctx.channel(), produceHar.getHeader(), topicPartition); - errorsConsumer.accept(Errors.INVALID_TOPIC_EXCEPTION); - return; - } - try { - final long beforeRecordsProcess = MathUtils.nowInNano(); - final MemoryRecords validRecords = - validateRecords(produceHar.getHeader().apiVersion(), topicPartition, records); - - validRecords.batches().forEach(batch->{ - if (batch.sizeInBytes() > kafkaConfig.getMaxMessageSize()) { - throw new RecordTooLargeException(String.format("Message batch size is %s " - + "in append to partition %s which exceeds the maximum configured size of %s .", - batch.sizeInBytes(), topicPartition, kafkaConfig.getMaxMessageSize())); - } - }); - - final CompletableFuture> topicFuture = - topicManager.getTopic(fullPartitionName); - if (topicFuture.isCompletedExceptionally()) { - topicFuture.exceptionally(e -> { - exceptionConsumer.accept(e); - return Optional.empty(); - }); - return; - } - if (topicFuture.isDone() && !topicFuture.getNow(Optional.empty()).isPresent()) { - errorsConsumer.accept(Errors.NOT_LEADER_FOR_PARTITION); - return; - } - - final Consumer> persistentTopicConsumer = persistentTopicOpt -> { - if (!persistentTopicOpt.isPresent()) { - errorsConsumer.accept(Errors.NOT_LEADER_FOR_PARTITION); - return; - } - - final EncodeRequest encodeRequest = EncodeRequest.get(validRecords); - if (entryFormatter instanceof KafkaMixedEntryFormatter) { - final ManagedLedger managedLedger = persistentTopicOpt.get().getManagedLedger(); - final long logEndOffset = MessageMetadataUtils.getLogEndOffset(managedLedger); - encodeRequest.setBaseOffset(logEndOffset); - } - - final EncodeResult encodeResult = entryFormatter.encode(encodeRequest); - encodeRequest.recycle(); - requestStats.getProduceEncodeStats().registerSuccessfulEvent( - MathUtils.elapsedNanos(beforeRecordsProcess), TimeUnit.NANOSECONDS); - startSendOperationForThrottling(encodeResult.getEncodedByteBuf().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, encodeResult, topicPartition, offsetConsumer, errorsConsumer); - }; - - if (topicFuture.isDone()) { - persistentTopicConsumer.accept(topicFuture.getNow(Optional.empty())); - } else { - // topic is not available now - pendingTopicFuturesMap - .computeIfAbsent(topicPartition, ignored -> - new PendingTopicFutures(requestStats)) - .addListener(topicFuture, persistentTopicConsumer, exceptionConsumer); - } - } catch (Exception e) { - log.error("[{}] Failed to handle produce request for {}", - ctx.channel(), topicPartition, e); - exceptionConsumer.accept(e); - } } @Override @@ -2681,55 +2519,6 @@ static AbstractResponse failedResponse(KafkaHeaderAndRequest requestHar, Throwab return requestHar.getRequest().getErrorResponse(((Integer) THROTTLE_TIME_MS.defaultValue), e); } - private static MemoryRecords validateRecords(short version, TopicPartition topicPartition, MemoryRecords records) { - if (version >= 3) { - Iterator iterator = records.batches().iterator(); - if (!iterator.hasNext()) { - throw new InvalidRecordException("Produce requests with version " + version + " must have at least " - + "one record batch"); - } - - MutableRecordBatch entry = iterator.next(); - if (entry.magic() != RecordBatch.MAGIC_VALUE_V2) { - throw new InvalidRecordException("Produce requests with version " + version + " are only allowed to " - + "contain record batches with magic version 2"); - } - - if (iterator.hasNext()) { - throw new InvalidRecordException("Produce requests with version " + version + " are only allowed to " - + "contain exactly one record batch"); - } - } - - int validBytesCount = 0; - for (RecordBatch batch : records.batches()) { - if (batch.magic() >= RecordBatch.MAGIC_VALUE_V2 && batch.baseOffset() != 0) { - throw new InvalidRecordException("The baseOffset of the record batch in the append to " - + topicPartition + " should be 0, but it is " + batch.baseOffset()); - } - - batch.ensureValid(); - validBytesCount += batch.sizeInBytes(); - } - - if (validBytesCount < 0) { - throw new CorruptRecordException("Cannot append record batch with illegal length " - + validBytesCount + " to log for " + topicPartition - + ". A possible cause is corrupted produce request."); - } - - MemoryRecords validRecords; - if (validBytesCount == records.sizeInBytes()) { - validRecords = records; - } else { - ByteBuffer validByteBuffer = records.buffer().duplicate(); - validByteBuffer.limit(validBytesCount); - validRecords = MemoryRecords.readableRecords(validByteBuffer); - } - - return validRecords; - } - @VisibleForTesting protected CompletableFuture authorize(AclOperation operation, Resource resource) { Session session = authenticator != null ? authenticator.session() : null; diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/PendingTopicFutures.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/PendingTopicFutures.java index c514fa0d24..21372f8f92 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/PendingTopicFutures.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/PendingTopicFutures.java @@ -55,7 +55,7 @@ private void registerQueueLatency(boolean success) { public void addListener(CompletableFuture> topicFuture, @NonNull Consumer> persistentTopicConsumer, - @NonNull Consumer exceptionConsumer) { + @NonNull CompletableFuture completableFuture) { if (count.compareAndSet(0, 1)) { // The first pending future comes currentTopicFuture = topicFuture.thenApply(persistentTopic -> { @@ -65,7 +65,7 @@ public void addListener(CompletableFuture> topicFuture return TopicThrowablePair.withTopic(persistentTopic); }).exceptionally(e -> { registerQueueLatency(false); - exceptionConsumer.accept(e.getCause()); + completableFuture.completeExceptionally(e.getCause()); count.decrementAndGet(); return TopicThrowablePair.withThrowable(e.getCause()); }); @@ -77,13 +77,13 @@ public void addListener(CompletableFuture> topicFuture persistentTopicConsumer.accept(topicThrowablePair.getPersistentTopicOpt()); } else { registerQueueLatency(false); - exceptionConsumer.accept(topicThrowablePair.getThrowable()); + completableFuture.completeExceptionally(topicThrowablePair.getThrowable()); } count.decrementAndGet(); return topicThrowablePair; }).exceptionally(e -> { registerQueueLatency(false); - exceptionConsumer.accept(e.getCause()); + completableFuture.completeExceptionally(e.getCause()); count.decrementAndGet(); return TopicThrowablePair.withThrowable(e.getCause()); }); diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/TenantContextManager.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/TenantContextManager.java index f5071bc382..0b63726dc5 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/TenantContextManager.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/TenantContextManager.java @@ -15,11 +15,13 @@ import io.streamnative.pulsar.handlers.kop.coordinator.group.GroupCoordinator; import io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionCoordinator; +import io.streamnative.pulsar.handlers.kop.storage.ReplicaManager; /** * Access Tenant level coordinators. */ public interface TenantContextManager { + /** * Access the GroupCoordinator for the current Tenant. * This method bootstraps a new GroupCoordinator if it is not started @@ -27,6 +29,7 @@ public interface TenantContextManager { * @return the GroupCoordinator */ GroupCoordinator getGroupCoordinator(String tenant); + /** * Access the TransactionCoordinator for the current Tenant. * This method bootstraps a new TransactionCoordinator if it is not started @@ -34,4 +37,12 @@ public interface TenantContextManager { * @return the TransactionCoordinator */ TransactionCoordinator getTransactionCoordinator(String tenant); + + /** + * Access the ReplicaManager for the current Tenant. + * This method bootstraps a new ReplicaManager if it is not started + * @param tenant + * @return the ReplicaManager + */ + ReplicaManager getReplicaManager(String tenant); } diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/AppendRecordsContext.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/AppendRecordsContext.java new file mode 100644 index 0000000000..546f5d1b0f --- /dev/null +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/AppendRecordsContext.java @@ -0,0 +1,72 @@ +/** + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.streamnative.pulsar.handlers.kop.storage; + +import io.netty.util.Recycler; +import io.streamnative.pulsar.handlers.kop.KafkaTopicManager; +import io.streamnative.pulsar.handlers.kop.PendingTopicFutures; +import io.streamnative.pulsar.handlers.kop.RequestStats; +import java.util.Map; +import java.util.function.Consumer; +import lombok.Getter; +import org.apache.kafka.common.TopicPartition; + +/** + * AppendRecordsContext is use for pass parameters to ReplicaManager, to avoid long parameter lists. + */ +@Getter +public class AppendRecordsContext { + private static final Recycler RECYCLER = new Recycler() { + protected AppendRecordsContext newObject(Handle handle) { + return new AppendRecordsContext(handle); + } + }; + + private final Recycler.Handle recyclerHandle; + private KafkaTopicManager topicManager; + private RequestStats requestStats; + private Consumer startSendOperationForThrottling; + private Consumer completeSendOperationForThrottling; + private Map pendingTopicFuturesMap; + + private AppendRecordsContext(Recycler.Handle recyclerHandle) { + this.recyclerHandle = recyclerHandle; + } + + // recycler and get for this object + public static AppendRecordsContext get(final KafkaTopicManager topicManager, + final RequestStats requestStats, + final Consumer startSendOperationForThrottling, + final Consumer completeSendOperationForThrottling, + final Map pendingTopicFuturesMap) { + AppendRecordsContext context = RECYCLER.get(); + context.topicManager = topicManager; + context.requestStats = requestStats; + context.startSendOperationForThrottling = startSendOperationForThrottling; + context.completeSendOperationForThrottling = completeSendOperationForThrottling; + context.pendingTopicFuturesMap = pendingTopicFuturesMap; + + return context; + } + + public void recycle() { + topicManager = null; + requestStats = null; + startSendOperationForThrottling = null; + completeSendOperationForThrottling = null; + pendingTopicFuturesMap = null; + recyclerHandle.recycle(this); + } + +} diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/PartitionLog.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/PartitionLog.java new file mode 100644 index 0000000000..ec3c695c50 --- /dev/null +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/PartitionLog.java @@ -0,0 +1,247 @@ +/** + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.streamnative.pulsar.handlers.kop.storage; + +import io.netty.buffer.ByteBuf; +import io.streamnative.pulsar.handlers.kop.KafkaServiceConfiguration; +import io.streamnative.pulsar.handlers.kop.KafkaTopicManager; +import io.streamnative.pulsar.handlers.kop.MessagePublishContext; +import io.streamnative.pulsar.handlers.kop.PendingTopicFutures; +import io.streamnative.pulsar.handlers.kop.RequestStats; +import io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionCoordinator; +import io.streamnative.pulsar.handlers.kop.format.EncodeRequest; +import io.streamnative.pulsar.handlers.kop.format.EncodeResult; +import io.streamnative.pulsar.handlers.kop.format.EntryFormatter; +import io.streamnative.pulsar.handlers.kop.format.KafkaMixedEntryFormatter; +import io.streamnative.pulsar.handlers.kop.utils.MessageMetadataUtils; +import java.nio.ByteBuffer; +import java.util.Iterator; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; +import lombok.AllArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.apache.bookkeeper.mledger.ManagedLedger; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.errors.CorruptRecordException; +import org.apache.kafka.common.errors.RecordTooLargeException; +import org.apache.kafka.common.protocol.Errors; +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.utils.Time; +import org.apache.pulsar.broker.service.persistent.PersistentTopic; +import org.apache.pulsar.common.naming.TopicName; + +/** + * An append-only log for storing messages. Mapping to Kafka Log.scala. + */ +@Slf4j +@AllArgsConstructor +public class PartitionLog { + private final KafkaServiceConfiguration kafkaConfig; + private final Time time; + private final TopicPartition topicPartition; + private final String namespacePrefix; + private final String fullPartitionName; + private final EntryFormatter entryFormatter; + private final Optional transactionCoordinator; + + /** + * Append this message to pulsar. + * + * @param records The log records to append + * @param version Inter-broker message protocol version + * @param appendRecordsContext See {@link AppendRecordsContext} + */ + public CompletableFuture appendRecords(final MemoryRecords records, + final short version, + final AppendRecordsContext appendRecordsContext) { + CompletableFuture appendFuture = new CompletableFuture<>(); + RequestStats requestStats = appendRecordsContext.getRequestStats(); + KafkaTopicManager topicManager = appendRecordsContext.getTopicManager(); + final long beforeRecordsProcess = time.nanoseconds(); + try { + MemoryRecords validRecords = validateRecords(version, fullPartitionName, records); + validRecords.batches().forEach(batch->{ + if (batch.sizeInBytes() > kafkaConfig.getMaxMessageSize()) { + throw new RecordTooLargeException(String.format("Message batch size is %s " + + "in append to partition %s which exceeds the maximum configured size of %s .", + batch.sizeInBytes(), topicPartition, kafkaConfig.getMaxMessageSize())); + } + }); + // Append Message into pulsar + final CompletableFuture> topicFuture = + topicManager.getTopic(fullPartitionName); + if (topicFuture.isCompletedExceptionally()) { + topicFuture.exceptionally(e -> { + appendFuture.completeExceptionally(e); + return Optional.empty(); + }); + return appendFuture; + } + if (topicFuture.isDone() && !topicFuture.getNow(Optional.empty()).isPresent()) { + appendFuture.completeExceptionally(Errors.NOT_LEADER_FOR_PARTITION.exception()); + return appendFuture; + } + final Consumer> persistentTopicConsumer = persistentTopicOpt -> { + if (!persistentTopicOpt.isPresent()) { + appendFuture.completeExceptionally(Errors.NOT_LEADER_FOR_PARTITION.exception()); + return; + } + // TODO: validateMessagesAndAssignOffsets here. + + final EncodeRequest encodeRequest = EncodeRequest.get(validRecords); + if (entryFormatter instanceof KafkaMixedEntryFormatter) { + final ManagedLedger managedLedger = persistentTopicOpt.get().getManagedLedger(); + final long logEndOffset = MessageMetadataUtils.getLogEndOffset(managedLedger); + encodeRequest.setBaseOffset(logEndOffset); + } + + final EncodeResult encodeResult = entryFormatter.encode(encodeRequest); + encodeRequest.recycle(); + requestStats.getProduceEncodeStats().registerSuccessfulEvent( + time.nanoseconds() - beforeRecordsProcess, TimeUnit.NANOSECONDS); + appendRecordsContext.getStartSendOperationForThrottling() + .accept(encodeResult.getEncodedByteBuf().readableBytes()); + + publishMessages(persistentTopicOpt, + appendFuture, + encodeResult, + topicPartition, + appendRecordsContext); + }; + + if (topicFuture.isDone()) { + persistentTopicConsumer.accept(topicFuture.getNow(Optional.empty())); + } else { + // topic is not available now + appendRecordsContext.getPendingTopicFuturesMap() + .computeIfAbsent(topicPartition, ignored -> + new PendingTopicFutures(requestStats)) + .addListener(topicFuture, persistentTopicConsumer, appendFuture); + } + } catch (Exception exception) { + log.error("Failed to handle produce request for {}", topicPartition, exception); + appendFuture.completeExceptionally(exception); + } + + return appendFuture; + } + + private void publishMessages(final Optional persistentTopicOpt, + final CompletableFuture appendFuture, + final EncodeResult encodeResult, + final TopicPartition topicPartition, + final AppendRecordsContext appendRecordsContext) { + final MemoryRecords records = encodeResult.getRecords(); + final int numMessages = encodeResult.getNumMessages(); + final ByteBuf byteBuf = encodeResult.getEncodedByteBuf(); + RequestStats requestStats = appendRecordsContext.getRequestStats(); + if (!persistentTopicOpt.isPresent()) { + encodeResult.recycle(); + // It will trigger a retry send of Kafka client + appendFuture.completeExceptionally(Errors.NOT_LEADER_FOR_PARTITION.exception()); + return; + } + PersistentTopic persistentTopic = persistentTopicOpt.get(); + if (persistentTopic.isSystemTopic()) { + encodeResult.recycle(); + log.error("Not support producing message to system topic: {}", persistentTopic); + appendFuture.completeExceptionally(Errors.INVALID_TOPIC_EXCEPTION.exception()); + return; + } + + appendRecordsContext.getTopicManager().registerProducerInPersistentTopic(fullPartitionName, persistentTopic); + + // collect metrics + encodeResult.updateProducerStats(topicPartition, requestStats, namespacePrefix); + + // publish + final CompletableFuture offsetFuture = new CompletableFuture<>(); + final long beforePublish = time.nanoseconds(); + persistentTopic.publishMessage(byteBuf, + MessagePublishContext.get(offsetFuture, persistentTopic, numMessages, System.nanoTime())); + final RecordBatch batch = records.batchIterator().next(); + offsetFuture.whenComplete((offset, e) -> { + appendRecordsContext.getCompleteSendOperationForThrottling().accept(byteBuf.readableBytes()); + encodeResult.recycle(); + if (e == null) { + if (batch.isTransactional()) { + transactionCoordinator.ifPresent(coordinator -> coordinator.addActivePidOffset( + TopicName.get(fullPartitionName), batch.producerId(), offset)); + } + requestStats.getMessagePublishStats().registerSuccessfulEvent( + time.nanoseconds() - beforePublish, TimeUnit.NANOSECONDS); + appendFuture.complete(offset); + } else { + log.error("publishMessages for topic partition: {} failed when write.", fullPartitionName, e); + requestStats.getMessagePublishStats().registerFailedEvent( + time.nanoseconds() - beforePublish, TimeUnit.NANOSECONDS); + appendFuture.completeExceptionally(Errors.KAFKA_STORAGE_ERROR.exception()); + } + }); + } + + private static MemoryRecords validateRecords(short version, String fullPartitionName, MemoryRecords records) { + if (version >= 3) { + Iterator iterator = records.batches().iterator(); + if (!iterator.hasNext()) { + throw new InvalidRecordException("Produce requests with version " + version + " must have at least " + + "one record batch"); + } + + MutableRecordBatch entry = iterator.next(); + if (entry.magic() != RecordBatch.MAGIC_VALUE_V2) { + throw new InvalidRecordException("Produce requests with version " + version + " are only allowed to " + + "contain record batches with magic version 2"); + } + + if (iterator.hasNext()) { + throw new InvalidRecordException("Produce requests with version " + version + " are only allowed to " + + "contain exactly one record batch"); + } + } + + int validBytesCount = 0; + for (RecordBatch batch : records.batches()) { + if (batch.magic() >= RecordBatch.MAGIC_VALUE_V2 && batch.baseOffset() != 0) { + throw new InvalidRecordException("The baseOffset of the record batch in the append to " + + fullPartitionName + " should be 0, but it is " + batch.baseOffset()); + } + + batch.ensureValid(); + validBytesCount += batch.sizeInBytes(); + } + + if (validBytesCount < 0) { + throw new CorruptRecordException("Cannot append record batch with illegal length " + + validBytesCount + " to log for " + fullPartitionName + + ". A possible cause is corrupted produce request."); + } + + MemoryRecords validRecords; + if (validBytesCount == records.sizeInBytes()) { + validRecords = records; + } else { + ByteBuffer validByteBuffer = records.buffer().duplicate(); + validByteBuffer.limit(validBytesCount); + validRecords = MemoryRecords.readableRecords(validByteBuffer); + } + + return validRecords; + } +} diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/PartitionLogManager.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/PartitionLogManager.java new file mode 100644 index 0000000000..1c2b04464d --- /dev/null +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/PartitionLogManager.java @@ -0,0 +1,58 @@ +/** + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.streamnative.pulsar.handlers.kop.storage; + +import com.google.common.collect.Maps; +import io.streamnative.pulsar.handlers.kop.KafkaServiceConfiguration; +import io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionCoordinator; +import io.streamnative.pulsar.handlers.kop.format.EntryFormatter; +import io.streamnative.pulsar.handlers.kop.utils.KopTopic; +import java.util.Map; +import java.util.Optional; +import lombok.AllArgsConstructor; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.utils.Time; + +/** + * Manage {@link PartitionLog}. + */ +@AllArgsConstructor +public class PartitionLogManager { + + private KafkaServiceConfiguration kafkaConfig; + private final Map logMap; + private final Optional transactionCoordinator; + private final EntryFormatter formatter; + private final Time time; + + public PartitionLogManager(KafkaServiceConfiguration kafkaConfig, + EntryFormatter entryFormatter, + Optional transactionCoordinator, + Time time) { + this.kafkaConfig = kafkaConfig; + this.logMap = Maps.newConcurrentMap(); + this.transactionCoordinator = transactionCoordinator; + this.formatter = entryFormatter; + this.time = time; + } + + public PartitionLog getLog(TopicPartition topicPartition, String namespacePrefix) { + String kopTopic = KopTopic.toString(topicPartition, namespacePrefix); + return logMap.computeIfAbsent(kopTopic, key -> + new PartitionLog(kafkaConfig, time, topicPartition, namespacePrefix, kopTopic, formatter, + this.transactionCoordinator) + ); + } +} + diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/ReplicaManager.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/ReplicaManager.java new file mode 100644 index 0000000000..3936dfcbd3 --- /dev/null +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/ReplicaManager.java @@ -0,0 +1,134 @@ +/** + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.streamnative.pulsar.handlers.kop.storage; + +import io.streamnative.pulsar.handlers.kop.DelayedProduceAndFetch; +import io.streamnative.pulsar.handlers.kop.KafkaServiceConfiguration; +import io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionCoordinator; +import io.streamnative.pulsar.handlers.kop.format.EntryFormatter; +import io.streamnative.pulsar.handlers.kop.utils.KopTopic; +import io.streamnative.pulsar.handlers.kop.utils.delayed.DelayedOperation; +import io.streamnative.pulsar.handlers.kop.utils.delayed.DelayedOperationKey; +import io.streamnative.pulsar.handlers.kop.utils.delayed.DelayedOperationPurgatory; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.BiConsumer; +import java.util.stream.Collectors; +import lombok.extern.slf4j.Slf4j; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.errors.InvalidTopicException; +import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.record.MemoryRecords; +import org.apache.kafka.common.requests.ProduceResponse; +import org.apache.kafka.common.utils.Time; + +/** + * Used to append records. Mapping to Kafka ReplicaManager.scala. + */ +@Slf4j +public class ReplicaManager { + private final PartitionLogManager logManager; + private final DelayedOperationPurgatory producePurgatory; + + public ReplicaManager(KafkaServiceConfiguration kafkaConfig, + Time time, + EntryFormatter entryFormatter, + Optional transactionCoordinator, + DelayedOperationPurgatory producePurgatory) { + this.logManager = new PartitionLogManager(kafkaConfig, entryFormatter, transactionCoordinator, time); + this.producePurgatory = producePurgatory; + } + + public PartitionLog getPartitionLog(TopicPartition topicPartition, String namespacePrefix) { + return logManager.getLog(topicPartition, namespacePrefix); + } + + public CompletableFuture> appendRecords( + final long timeout, + final boolean internalTopicsAllowed, + final short version, + final String namespacePrefix, + final Map entriesPerPartition, + final AppendRecordsContext appendRecordsContext) { + CompletableFuture> completableFuture = + new CompletableFuture<>(); + final AtomicInteger topicPartitionNum = new AtomicInteger(entriesPerPartition.size()); + final Map responseMap = new ConcurrentHashMap<>(); + + Runnable complete = () -> { + topicPartitionNum.set(0); + if (completableFuture.isDone()) { + // It may be triggered again in DelayedProduceAndFetch + return; + } + // add the topicPartition with timeout error if it's not existed in responseMap + entriesPerPartition.keySet().forEach(topicPartition -> { + if (!responseMap.containsKey(topicPartition)) { + responseMap.put(topicPartition, new ProduceResponse.PartitionResponse(Errors.REQUEST_TIMED_OUT)); + } + }); + if (log.isDebugEnabled()) { + log.debug("Complete handle appendRecords."); + } + completableFuture.complete(responseMap); + }; + BiConsumer addPartitionResponse = + (topicPartition, response) -> { + responseMap.put(topicPartition, response); + // reset topicPartitionNum + int restTopicPartitionNum = topicPartitionNum.decrementAndGet(); + if (restTopicPartitionNum < 0) { + return; + } + if (restTopicPartitionNum == 0) { + complete.run(); + } + }; + entriesPerPartition.forEach((topicPartition, memoryRecords) -> { + String fullPartitionName = KopTopic.toString(topicPartition, namespacePrefix); + // reject appending to internal topics if it is not allowed + if (!internalTopicsAllowed && KopTopic.isInternalTopic(fullPartitionName)) { + addPartitionResponse.accept(topicPartition, new ProduceResponse.PartitionResponse( + Errors.forException(new InvalidTopicException( + String.format("Cannot append to internal topic %s", topicPartition.topic()))))); + } else { + PartitionLog partitionLog = getPartitionLog(topicPartition, namespacePrefix); + partitionLog.appendRecords(memoryRecords, version, appendRecordsContext) + .thenAccept(offset -> addPartitionResponse.accept(topicPartition, + new ProduceResponse.PartitionResponse(Errors.NONE, offset, -1L, -1L))) + .exceptionally(ex -> { + addPartitionResponse.accept(topicPartition, + new ProduceResponse.PartitionResponse(Errors.forException(ex.getCause()))); + return null; + }); + } + }); + // delay produce + if (timeout <= 0) { + complete.run(); + } else { + List delayedCreateKeys = + entriesPerPartition.keySet().stream() + .map(DelayedOperationKey.TopicPartitionOperationKey::new).collect(Collectors.toList()); + DelayedProduceAndFetch delayedProduce = new DelayedProduceAndFetch(timeout, topicPartitionNum, complete); + producePurgatory.tryCompleteElseWatch(delayedProduce, delayedCreateKeys); + } + return completableFuture; + } + +} diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/package-info.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/package-info.java new file mode 100644 index 0000000000..ad5d79b79c --- /dev/null +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/storage/package-info.java @@ -0,0 +1,14 @@ +/** + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.streamnative.pulsar.handlers.kop.storage; \ No newline at end of file diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/KopTopic.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/KopTopic.java index 1f5f4a71dd..fe75a7f654 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/KopTopic.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/KopTopic.java @@ -13,11 +13,14 @@ */ package io.streamnative.pulsar.handlers.kop.utils; +import static org.apache.kafka.common.internals.Topic.GROUP_METADATA_TOPIC_NAME; +import static org.apache.kafka.common.internals.Topic.TRANSACTION_STATE_TOPIC_NAME; import static org.apache.pulsar.common.naming.TopicName.PARTITIONED_TOPIC_SUFFIX; import io.streamnative.pulsar.handlers.kop.exceptions.KoPTopicException; import lombok.Getter; import org.apache.kafka.common.TopicPartition; +import org.apache.pulsar.common.naming.TopicName; /** * KopTopic maintains two topic name, one is the original topic name, the other is the full topic name used in Pulsar. @@ -103,4 +106,10 @@ public static String toString(String topic, int partition, String namespacePrefi return (new KopTopic(topic, namespacePrefix)).getPartitionName(partition); } + public static boolean isInternalTopic(final String fullTopicName) { + String partitionedTopicName = TopicName.get(fullTopicName).getPartitionedTopicName(); + return partitionedTopicName.endsWith("/" + GROUP_METADATA_TOPIC_NAME) + || partitionedTopicName.endsWith("/" + TRANSACTION_STATE_TOPIC_NAME); + } + } diff --git a/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/PendingTopicFuturesTest.java b/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/PendingTopicFuturesTest.java index 7131599cff..053c7bd42a 100644 --- a/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/PendingTopicFuturesTest.java +++ b/kafka-impl/src/test/java/io/streamnative/pulsar/handlers/kop/PendingTopicFuturesTest.java @@ -64,7 +64,8 @@ void testNormalComplete() throws ExecutionException, InterruptedException { for (int i = 0; i < 10; i++) { final int index = i; - pendingTopicFutures.addListener(topicFuture, ignored -> completedIndexes.add(index), e -> {}); + pendingTopicFutures.addListener( + topicFuture, ignored -> completedIndexes.add(index), new CompletableFuture<>()); changesOfPendingCount.add(pendingTopicFutures.size()); sleep(234); } @@ -95,7 +96,12 @@ void testExceptionalComplete() throws ExecutionException, InterruptedException { final List changesOfPendingCount = new ArrayList<>(); for (int i = 0; i < 10; i++) { - pendingTopicFutures.addListener(topicFuture, topic -> {}, e -> exceptionMessages.add(e.getMessage())); + CompletableFuture longCompletableFuture = new CompletableFuture<>(); + longCompletableFuture.exceptionally(ex -> { + exceptionMessages.add(ex.getMessage()); + return null; + }); + pendingTopicFutures.addListener(topicFuture, topic -> {}, longCompletableFuture); changesOfPendingCount.add(pendingTopicFutures.size()); sleep(200); } diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KopProtocolHandlerTestBase.java b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KopProtocolHandlerTestBase.java index e32de60c8d..e37f4011d1 100644 --- a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KopProtocolHandlerTestBase.java +++ b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KopProtocolHandlerTestBase.java @@ -26,6 +26,7 @@ import io.streamnative.pulsar.handlers.kop.coordinator.group.GroupCoordinator; import io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionCoordinator; import io.streamnative.pulsar.handlers.kop.stats.NullStatsLogger; +import io.streamnative.pulsar.handlers.kop.storage.ReplicaManager; import io.streamnative.pulsar.handlers.kop.utils.MetadataUtils; import java.io.Closeable; import java.io.IOException; @@ -768,6 +769,8 @@ public KafkaRequestHandler newRequestHandler() throws Exception { final GroupCoordinator groupCoordinator = handler.getGroupCoordinator(conf.getKafkaMetadataTenant()); final TransactionCoordinator transactionCoordinator = handler.getTransactionCoordinator(conf.getKafkaMetadataTenant()); + final ReplicaManager replicaManager = + handler.getReplicaManager(conf.getKafkaMetadataTenant()); return ((KafkaChannelInitializer) handler.getChannelInitializerMap().entrySet().iterator().next().getValue()) .newCnx(new TenantContextManager() { @@ -780,6 +783,11 @@ public GroupCoordinator getGroupCoordinator(String tenant) { public TransactionCoordinator getTransactionCoordinator(String tenant) { return transactionCoordinator; } + + @Override + public ReplicaManager getReplicaManager(String tenant) { + return replicaManager; + } }, NullStatsLogger.INSTANCE); } }