From 16f1e5e9639cd7b3b4ae908f9056083fe8017dcd Mon Sep 17 00:00:00 2001 From: Demogorgon314 Date: Thu, 26 May 2022 13:31:52 +0800 Subject: [PATCH 1/4] Restore the support for Kafka clients 0.9 and 0.10 --- kafka-client-2-0-shade/pom.xml | 68 +++++++++++++++++++ kafka-impl/pom.xml | 5 ++ .../handlers/kop/KafkaCommandDecoder.java | 14 ++++ .../handlers/kop/KafkaRequestHandler.java | 66 ++++++++++++++++-- .../handlers/kop/utils/KafkaRequestUtils.java | 10 +++ pom.xml | 7 ++ .../compatibility/BasicEndToEndTestBase.java | 1 - .../saslplain/SaslPlainEndToEndTest.java | 1 - 8 files changed, 163 insertions(+), 9 deletions(-) create mode 100644 kafka-client-2-0-shade/pom.xml diff --git a/kafka-client-2-0-shade/pom.xml b/kafka-client-2-0-shade/pom.xml new file mode 100644 index 0000000000..0366d966d1 --- /dev/null +++ b/kafka-client-2-0-shade/pom.xml @@ -0,0 +1,68 @@ + + + + + pulsar-protocol-handler-kafka-parent + io.streamnative.pulsar.handlers + 2.11.0-SNAPSHOT + + 4.0.0 + + kafka-client-2-0-shade + StreamNative :: Pulsar Protocol Handler :: Kafka Client 2.0.0 shade + The Kafka client 2.0.0 shade package + + + + org.apache.kafka + kafka-clients + 2.0.0 + + + + + + + org.apache.maven.plugins + maven-shade-plugin + ${maven-shade-plugin.version} + + + package + + shade + + + + + org.apache.kafka + org.apache.kafka200 + + + org.apache.kafka + org.apache.kafka200 + + + + + + + + + \ No newline at end of file diff --git a/kafka-impl/pom.xml b/kafka-impl/pom.xml index 755c81a4dd..2e2a96f43f 100644 --- a/kafka-impl/pom.xml +++ b/kafka-impl/pom.xml @@ -55,6 +55,11 @@ kafka-clients + + io.streamnative.pulsar.handlers + kafka-client-2-0-shade + + org.projectlombok lombok diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaCommandDecoder.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaCommandDecoder.java index 048a38753c..f5d5b38680 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaCommandDecoder.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaCommandDecoder.java @@ -154,6 +154,20 @@ protected KafkaHeaderAndRequest byteBufToRequest(ByteBuf msg, } } + protected org.apache.kafka200.common.requests.ListOffsetRequest byteBufToListOffsetRequestV0(ByteBuf buf) { + checkArgument(buf.readableBytes() > 0); + ByteBuffer nio = buf.nioBuffer(); + org.apache.kafka200.common.requests.RequestHeader header = + org.apache.kafka200.common.requests.RequestHeader.parse(nio); + short apiVersion = header.apiVersion(); + org.apache.kafka200.common.protocol.ApiKeys apiKey = header.apiKey(); + org.apache.kafka200.common.protocol.types.Struct struct = apiKey.parseRequest(apiVersion, nio); + org.apache.kafka200.common.requests.AbstractRequest body = + org.apache.kafka200.common.requests.AbstractRequest.parseRequest(apiKey, apiVersion, struct); + + return (org.apache.kafka200.common.requests.ListOffsetRequest) body; + } + protected static ByteBuf responseToByteBuf(AbstractResponse response, KafkaHeaderAndRequest request) { try (KafkaHeaderAndResponse kafkaHeaderAndResponse = KafkaHeaderAndResponse.responseForRequest(request, response)) { 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 d6836f8c27..4ab2aab5e9 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 @@ -1076,8 +1076,7 @@ protected void handleOffsetFetchRequest(KafkaHeaderAndRequest offsetFetch, }); } - private CompletableFuture> fetchOffset(String topicName, ListOffsetRequest.PartitionData pd) { - Long timestamp = pd.timestamp; + private CompletableFuture> fetchOffset(String topicName, long timestamp) { CompletableFuture> partitionData = new CompletableFuture<>(); topicManager.getTopic(topicName).whenComplete((perTopicOpt, t) -> { @@ -1258,7 +1257,7 @@ private void handleListOffsetRequestV1AndAbove(KafkaHeaderAndRequest listOffset, completeOne.run(); return; } - responseData.put(topic, fetchOffset(fullPartitionName, times)); + responseData.put(topic, fetchOffset(fullPartitionName, times.timestamp)); completeOne.run(); } ); @@ -1271,10 +1270,63 @@ private void handleListOffsetRequestV1AndAbove(KafkaHeaderAndRequest listOffset, // https://cfchou.github.io/blog/2015/04/23/a-closer-look-at-kafka-offsetrequest/ through web.archive.org private void handleListOffsetRequestV0(KafkaHeaderAndRequest listOffset, CompletableFuture resultFuture) { - log.error("{} ListOffset v0 is not supported", this); - resultFuture.complete(listOffset - .getRequest() - .getErrorResponse(new Exception("V0 not supported"))); + org.apache.kafka200.common.requests.ListOffsetRequest request = + byteBufToListOffsetRequestV0(listOffset.getBuffer()); + + Map>> responseData = + Maps.newConcurrentMap(); + if (request.offsetData().size() == 0) { + resultFuture.complete(new ListOffsetResponse(Collections.emptyMap())); + return; + } + AtomicInteger partitions = new AtomicInteger(request.offsetData().size()); + Runnable completeOne = () -> { + if (partitions.decrementAndGet() == 0) { + waitResponseDataComplete(resultFuture, responseData, true); + } + }; + // in v0, the iterator is offsetData, + // in v1, the iterator is partitionTimestamps, + if (log.isDebugEnabled()) { + log.debug("received a v0 listOffset: {}", request.toString(true)); + } + String namespacePrefix = currentNamespacePrefix(); + KafkaRequestUtils.LegacyUtils.forEachListOffsetRequest(request, topic -> times -> maxNumOffsets -> { + TopicPartition topicV1 = new TopicPartition(topic.topic(), topic.partition()); + String fullPartitionName = KopTopic.toString(topicV1, namespacePrefix); + + authorize(AclOperation.DESCRIBE, Resource.of(ResourceType.TOPIC, fullPartitionName)) + .whenComplete((isAuthorized, ex) -> { + if (ex != null) { + log.error("Describe topic authorize failed, topic - {}. {}", + fullPartitionName, ex.getMessage()); + responseData.put(topicV1, CompletableFuture.completedFuture( + Pair.of(Errors.TOPIC_AUTHORIZATION_FAILED, null))); + completeOne.run(); + return; + } + if (!isAuthorized) { + responseData.put(topicV1, CompletableFuture.completedFuture( + Pair.of(Errors.TOPIC_AUTHORIZATION_FAILED, null))); + completeOne.run(); + return; + } + + CompletableFuture> partitionData; + // num_num_offsets > 1 is not handled for now, returning an error + if (maxNumOffsets > 1) { + log.warn("request is asking for multiples offsets for {}, not supported for now", + fullPartitionName); + partitionData = new CompletableFuture<>(); + partitionData.complete(Pair.of(Errors.UNKNOWN_SERVER_ERROR, null)); + } + + partitionData = fetchOffset(fullPartitionName, times); + responseData.put(topicV1, partitionData); + completeOne.run(); + }); + + }); } // get offset from underline managedLedger diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/KafkaRequestUtils.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/KafkaRequestUtils.java index 5020f94794..7ff9049c79 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/KafkaRequestUtils.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/KafkaRequestUtils.java @@ -16,6 +16,8 @@ import io.streamnative.pulsar.handlers.kop.offset.OffsetAndMetadata; import java.util.Optional; import java.util.function.BiConsumer; +import java.util.function.Consumer; +import java.util.function.Function; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.requests.CreatePartitionsRequest; import org.apache.kafka.common.requests.ListOffsetRequest; @@ -44,6 +46,14 @@ public static long getOffset(TxnOffsetCommitRequest.CommittedOffset committedOff public static class LegacyUtils { + public static void forEachListOffsetRequest( + org.apache.kafka200.common.requests.ListOffsetRequest request, + Function>> function) { + request.offsetData().forEach((topicPartition, partitionData) -> { + function.apply(topicPartition).apply(partitionData.timestamp).accept(partitionData.maxNumOffsets); + }); + } + // V2 adds retention time to the request and V5 removes retention time public static long getRetentionTime(OffsetCommitRequest request) { return request.retentionTime(); diff --git a/pom.xml b/pom.xml index d3d513c724..3cc13aa289 100644 --- a/pom.xml +++ b/pom.xml @@ -87,6 +87,7 @@ kafka-payload-processor-shaded kafka-payload-processor-shaded-tests test-listener + kafka-client-2-0-shade @@ -98,6 +99,12 @@ ${kafka.version} + + io.streamnative.pulsar.handlers + kafka-client-2-0-shade + ${project.version} + + ${pulsar.group.id} pulsar-broker diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/compatibility/BasicEndToEndTestBase.java b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/compatibility/BasicEndToEndTestBase.java index 9c41122832..5b0838b0a9 100644 --- a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/compatibility/BasicEndToEndTestBase.java +++ b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/compatibility/BasicEndToEndTestBase.java @@ -49,7 +49,6 @@ public class BasicEndToEndTestBase extends KopProtocolHandlerTestBase { protected Map kafkaClientFactories = Arrays.stream(KafkaVersion.values()) - .filter(k -> k != KafkaVersion.KAFKA_0_9_0_0 && k != KafkaVersion.KAFKA_0_10_0_0) .collect(Collectors.toMap( version -> version, version -> { diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/compatibility/saslplain/SaslPlainEndToEndTest.java b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/compatibility/saslplain/SaslPlainEndToEndTest.java index f2c9d23468..381efe9c06 100644 --- a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/compatibility/saslplain/SaslPlainEndToEndTest.java +++ b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/compatibility/saslplain/SaslPlainEndToEndTest.java @@ -79,7 +79,6 @@ public class SaslPlainEndToEndTest extends KopProtocolHandlerTestBase { private File jaasConfigFile; protected Map kafkaClientFactories = Arrays.stream(KafkaVersion.values()) - .filter(k -> k != KafkaVersion.KAFKA_0_9_0_0 && k != KafkaVersion.KAFKA_0_10_0_0) .collect(Collectors.toMap( version -> version, version -> { From 81906828d9e3148c42836601c254a2d2207b4a20 Mon Sep 17 00:00:00 2001 From: Demogorgon314 Date: Thu, 26 May 2022 13:35:48 +0800 Subject: [PATCH 2/4] Remove unused code --- kafka-client-2-0-shade/pom.xml | 4 ---- 1 file changed, 4 deletions(-) diff --git a/kafka-client-2-0-shade/pom.xml b/kafka-client-2-0-shade/pom.xml index 0366d966d1..7654831023 100644 --- a/kafka-client-2-0-shade/pom.xml +++ b/kafka-client-2-0-shade/pom.xml @@ -54,10 +54,6 @@ org.apache.kafka org.apache.kafka200 - - org.apache.kafka - org.apache.kafka200 - From 26127c509e8d46596c854fd043cc8db2934a493a Mon Sep 17 00:00:00 2001 From: Demogorgon314 Date: Thu, 26 May 2022 17:57:52 +0800 Subject: [PATCH 3/4] Copy ListOffsetRequest to KoP --- kafka-client-2-0-shade/pom.xml | 64 --- kafka-impl/pom.xml | 5 - .../handlers/kop/KafkaCommandDecoder.java | 13 +- .../handlers/kop/KafkaRequestHandler.java | 12 +- .../handlers/kop/utils/KafkaRequestUtils.java | 5 +- .../common/requests/ListOffsetRequestV0.java | 373 ++++++++++++++++++ pom.xml | 7 - 7 files changed, 386 insertions(+), 93 deletions(-) delete mode 100644 kafka-client-2-0-shade/pom.xml create mode 100644 kafka-impl/src/main/java/org/apache/kafka/common/requests/ListOffsetRequestV0.java diff --git a/kafka-client-2-0-shade/pom.xml b/kafka-client-2-0-shade/pom.xml deleted file mode 100644 index 7654831023..0000000000 --- a/kafka-client-2-0-shade/pom.xml +++ /dev/null @@ -1,64 +0,0 @@ - - - - - pulsar-protocol-handler-kafka-parent - io.streamnative.pulsar.handlers - 2.11.0-SNAPSHOT - - 4.0.0 - - kafka-client-2-0-shade - StreamNative :: Pulsar Protocol Handler :: Kafka Client 2.0.0 shade - The Kafka client 2.0.0 shade package - - - - org.apache.kafka - kafka-clients - 2.0.0 - - - - - - - org.apache.maven.plugins - maven-shade-plugin - ${maven-shade-plugin.version} - - - package - - shade - - - - - org.apache.kafka - org.apache.kafka200 - - - - - - - - - \ No newline at end of file diff --git a/kafka-impl/pom.xml b/kafka-impl/pom.xml index 2e2a96f43f..755c81a4dd 100644 --- a/kafka-impl/pom.xml +++ b/kafka-impl/pom.xml @@ -55,11 +55,6 @@ kafka-clients - - io.streamnative.pulsar.handlers - kafka-client-2-0-shade - - org.projectlombok lombok diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaCommandDecoder.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaCommandDecoder.java index f5d5b38680..baa9fd2d02 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaCommandDecoder.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaCommandDecoder.java @@ -43,6 +43,7 @@ import org.apache.kafka.common.requests.AbstractResponse; import org.apache.kafka.common.requests.ApiVersionsRequest; import org.apache.kafka.common.requests.KopResponseUtils; +import org.apache.kafka.common.requests.ListOffsetRequestV0; import org.apache.kafka.common.requests.RequestHeader; import org.apache.kafka.common.requests.ResponseCallbackWrapper; import org.apache.kafka.common.requests.ResponseHeader; @@ -154,18 +155,12 @@ protected KafkaHeaderAndRequest byteBufToRequest(ByteBuf msg, } } - protected org.apache.kafka200.common.requests.ListOffsetRequest byteBufToListOffsetRequestV0(ByteBuf buf) { + protected ListOffsetRequestV0 byteBufToListOffsetRequestV0(ByteBuf buf) { checkArgument(buf.readableBytes() > 0); ByteBuffer nio = buf.nioBuffer(); - org.apache.kafka200.common.requests.RequestHeader header = - org.apache.kafka200.common.requests.RequestHeader.parse(nio); + RequestHeader header = RequestHeader.parse(nio); short apiVersion = header.apiVersion(); - org.apache.kafka200.common.protocol.ApiKeys apiKey = header.apiKey(); - org.apache.kafka200.common.protocol.types.Struct struct = apiKey.parseRequest(apiVersion, nio); - org.apache.kafka200.common.requests.AbstractRequest body = - org.apache.kafka200.common.requests.AbstractRequest.parseRequest(apiKey, apiVersion, struct); - - return (org.apache.kafka200.common.requests.ListOffsetRequest) body; + return ListOffsetRequestV0.parse(nio, apiVersion); } protected static ByteBuf responseToByteBuf(AbstractResponse response, KafkaHeaderAndRequest request) { 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 4ab2aab5e9..448fabaae5 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 @@ -140,6 +140,7 @@ import org.apache.kafka.common.requests.LeaveGroupRequest; import org.apache.kafka.common.requests.ListGroupsRequest; import org.apache.kafka.common.requests.ListOffsetRequest; +import org.apache.kafka.common.requests.ListOffsetRequestV0; import org.apache.kafka.common.requests.ListOffsetResponse; import org.apache.kafka.common.requests.MetadataRequest; import org.apache.kafka.common.requests.MetadataResponse.PartitionMetadata; @@ -1270,7 +1271,7 @@ private void handleListOffsetRequestV1AndAbove(KafkaHeaderAndRequest listOffset, // https://cfchou.github.io/blog/2015/04/23/a-closer-look-at-kafka-offsetrequest/ through web.archive.org private void handleListOffsetRequestV0(KafkaHeaderAndRequest listOffset, CompletableFuture resultFuture) { - org.apache.kafka200.common.requests.ListOffsetRequest request = + ListOffsetRequestV0 request = byteBufToListOffsetRequestV0(listOffset.getBuffer()); Map>> responseData = @@ -1292,21 +1293,20 @@ private void handleListOffsetRequestV0(KafkaHeaderAndRequest listOffset, } String namespacePrefix = currentNamespacePrefix(); KafkaRequestUtils.LegacyUtils.forEachListOffsetRequest(request, topic -> times -> maxNumOffsets -> { - TopicPartition topicV1 = new TopicPartition(topic.topic(), topic.partition()); - String fullPartitionName = KopTopic.toString(topicV1, namespacePrefix); + String fullPartitionName = KopTopic.toString(topic, namespacePrefix); authorize(AclOperation.DESCRIBE, Resource.of(ResourceType.TOPIC, fullPartitionName)) .whenComplete((isAuthorized, ex) -> { if (ex != null) { log.error("Describe topic authorize failed, topic - {}. {}", fullPartitionName, ex.getMessage()); - responseData.put(topicV1, CompletableFuture.completedFuture( + responseData.put(topic, CompletableFuture.completedFuture( Pair.of(Errors.TOPIC_AUTHORIZATION_FAILED, null))); completeOne.run(); return; } if (!isAuthorized) { - responseData.put(topicV1, CompletableFuture.completedFuture( + responseData.put(topic, CompletableFuture.completedFuture( Pair.of(Errors.TOPIC_AUTHORIZATION_FAILED, null))); completeOne.run(); return; @@ -1322,7 +1322,7 @@ private void handleListOffsetRequestV0(KafkaHeaderAndRequest listOffset, } partitionData = fetchOffset(fullPartitionName, times); - responseData.put(topicV1, partitionData); + responseData.put(topic, partitionData); completeOne.run(); }); diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/KafkaRequestUtils.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/KafkaRequestUtils.java index 7ff9049c79..6c04cd0212 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/KafkaRequestUtils.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/KafkaRequestUtils.java @@ -21,6 +21,7 @@ import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.requests.CreatePartitionsRequest; import org.apache.kafka.common.requests.ListOffsetRequest; +import org.apache.kafka.common.requests.ListOffsetRequestV0; import org.apache.kafka.common.requests.OffsetCommitRequest; import org.apache.kafka.common.requests.TxnOffsetCommitRequest; @@ -47,8 +48,8 @@ public static long getOffset(TxnOffsetCommitRequest.CommittedOffset committedOff public static class LegacyUtils { public static void forEachListOffsetRequest( - org.apache.kafka200.common.requests.ListOffsetRequest request, - Function>> function) { + ListOffsetRequestV0 request, + Function>> function) { request.offsetData().forEach((topicPartition, partitionData) -> { function.apply(topicPartition).apply(partitionData.timestamp).accept(partitionData.maxNumOffsets); }); diff --git a/kafka-impl/src/main/java/org/apache/kafka/common/requests/ListOffsetRequestV0.java b/kafka-impl/src/main/java/org/apache/kafka/common/requests/ListOffsetRequestV0.java new file mode 100644 index 0000000000..e972a9db7f --- /dev/null +++ b/kafka-impl/src/main/java/org/apache/kafka/common/requests/ListOffsetRequestV0.java @@ -0,0 +1,373 @@ +/** + * 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 org.apache.kafka.common.requests; + +import static org.apache.kafka.common.protocol.CommonFields.PARTITION_ID; +import static org.apache.kafka.common.protocol.CommonFields.TOPIC_NAME; +import static org.apache.kafka.common.protocol.types.Type.INT32; +import static org.apache.kafka.common.protocol.types.Type.INT64; +import static org.apache.kafka.common.protocol.types.Type.INT8; + +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.protocol.types.ArrayOf; +import org.apache.kafka.common.protocol.types.Field; +import org.apache.kafka.common.protocol.types.Schema; +import org.apache.kafka.common.protocol.types.Struct; + +public class ListOffsetRequestV0 extends AbstractRequest { + public static final long EARLIEST_TIMESTAMP = -2L; + public static final long LATEST_TIMESTAMP = -1L; + + public static final int CONSUMER_REPLICA_ID = -1; + public static final int DEBUGGING_REPLICA_ID = -2; + + private static final String REPLICA_ID_KEY_NAME = "replica_id"; + private static final String ISOLATION_LEVEL_KEY_NAME = "isolation_level"; + private static final String TOPICS_KEY_NAME = "topics"; + + // topic level field names + private static final String PARTITIONS_KEY_NAME = "partitions"; + + // partition level field names + private static final String TIMESTAMP_KEY_NAME = "timestamp"; + private static final String MAX_NUM_OFFSETS_KEY_NAME = "max_num_offsets"; + + private static final Schema LIST_OFFSET_REQUEST_PARTITION_V0 = new Schema( + PARTITION_ID, + new Field(TIMESTAMP_KEY_NAME, INT64, "Timestamp."), + new Field(MAX_NUM_OFFSETS_KEY_NAME, INT32, "Maximum offsets to return.")); + private static final Schema LIST_OFFSET_REQUEST_PARTITION_V1 = new Schema( + PARTITION_ID, + new Field(TIMESTAMP_KEY_NAME, INT64, "The target timestamp for the partition.")); + + private static final Schema LIST_OFFSET_REQUEST_TOPIC_V0 = new Schema( + TOPIC_NAME, + new Field(PARTITIONS_KEY_NAME, new ArrayOf(LIST_OFFSET_REQUEST_PARTITION_V0), + "Partitions to list offset.")); + private static final Schema LIST_OFFSET_REQUEST_TOPIC_V1 = new Schema( + TOPIC_NAME, + new Field(PARTITIONS_KEY_NAME, new ArrayOf(LIST_OFFSET_REQUEST_PARTITION_V1), + "Partitions to list offset.")); + + private static final Schema LIST_OFFSET_REQUEST_V0 = new Schema( + new Field(REPLICA_ID_KEY_NAME, INT32, "Broker id of the follower. For normal consumers, use -1."), + new Field(TOPICS_KEY_NAME, new ArrayOf(LIST_OFFSET_REQUEST_TOPIC_V0), "Topics to list offsets.")); + private static final Schema LIST_OFFSET_REQUEST_V1 = new Schema( + new Field(REPLICA_ID_KEY_NAME, INT32, "Broker id of the follower. For normal consumers, use -1."), + new Field(TOPICS_KEY_NAME, new ArrayOf(LIST_OFFSET_REQUEST_TOPIC_V1), "Topics to list offsets.")); + + private static final Schema LIST_OFFSET_REQUEST_V2 = new Schema( + new Field(REPLICA_ID_KEY_NAME, INT32, "Broker id of the follower. For normal consumers, use -1."), + new Field(ISOLATION_LEVEL_KEY_NAME, INT8, + "This setting controls the visibility of transactional records. " + + "Using READ_UNCOMMITTED (isolation_level = 0) makes all records visible. With READ_COMMITTED " + + "(isolation_level = 1), non-transactional and COMMITTED transactional records are visible. " + + "To be more concrete, READ_COMMITTED returns all data from offsets smaller than the current " + + "LSO (last stable offset), and enables the inclusion of the list of aborted transactions in the " + + "result, which allows consumers to discard ABORTED transactional records"), + new Field(TOPICS_KEY_NAME, new ArrayOf(LIST_OFFSET_REQUEST_TOPIC_V1), "Topics to list offsets.")); + + /** + * The version number is bumped to indicate that on quota violation brokers send out responses before throttling. + */ + private static final Schema LIST_OFFSET_REQUEST_V3 = LIST_OFFSET_REQUEST_V2; + + public static Schema[] schemaVersions() { + return new Schema[] {LIST_OFFSET_REQUEST_V0, LIST_OFFSET_REQUEST_V1, LIST_OFFSET_REQUEST_V2, + LIST_OFFSET_REQUEST_V3}; + } + + private final int replicaId; + private final IsolationLevel isolationLevel; + private final Map offsetData; + private final Map partitionTimestamps; + private final Set duplicatePartitions; + + public static class Builder extends AbstractRequest.Builder { + private final int replicaId; + private final IsolationLevel isolationLevel; + private Map offsetData = null; + private Map partitionTimestamps = null; + + public static ListOffsetRequestV0.Builder forReplica(short allowedVersion, int replicaId) { + return new ListOffsetRequestV0.Builder((short) 0, allowedVersion, replicaId, + IsolationLevel.READ_UNCOMMITTED); + } + + public static ListOffsetRequestV0.Builder forConsumer(boolean requireTimestamp, IsolationLevel isolationLevel) { + short minVersion = 0; + if (isolationLevel == IsolationLevel.READ_COMMITTED) { + minVersion = 2; + } else if (requireTimestamp){ + minVersion = 1; + } + return new ListOffsetRequestV0.Builder(minVersion, + ApiKeys.LIST_OFFSETS.latestVersion(), + CONSUMER_REPLICA_ID, isolationLevel); + } + + private Builder(short oldestAllowedVersion, short latestAllowedVersion, int replicaId, + IsolationLevel isolationLevel) { + super(ApiKeys.LIST_OFFSETS, oldestAllowedVersion, latestAllowedVersion); + this.replicaId = replicaId; + this.isolationLevel = isolationLevel; + } + + public ListOffsetRequestV0.Builder setOffsetData(Map offsetData) { + this.offsetData = offsetData; + return this; + } + + public ListOffsetRequestV0.Builder setTargetTimes(Map partitionTimestamps) { + this.partitionTimestamps = partitionTimestamps; + return this; + } + + @Override + public ListOffsetRequestV0 build(short version) { + if (version == 0) { + if (offsetData == null) { + if (partitionTimestamps == null) { + throw new RuntimeException("Must set partitionTimestamps or offsetData when creating a v0 " + + "ListOffsetRequest"); + } else { + offsetData = new HashMap<>(); + for (Map.Entry entry: partitionTimestamps.entrySet()) { + offsetData.put(entry.getKey(), + new ListOffsetRequestV0.PartitionData(entry.getValue(), 1)); + } + this.partitionTimestamps = null; + } + } + } else { + if (offsetData != null) { + throw new RuntimeException("Cannot create a v" + version + " ListOffsetRequest with v0 " + + "PartitionData."); + } else if (partitionTimestamps == null) { + throw new RuntimeException("Must set partitionTimestamps when creating a v" + + version + " ListOffsetRequest"); + } + } + Map m = (version == 0) ? offsetData : partitionTimestamps; + return new ListOffsetRequestV0(replicaId, m, isolationLevel, version); + } + + @Override + public String toString() { + StringBuilder bld = new StringBuilder(); + bld.append("(type=ListOffsetRequest") + .append(", replicaId=").append(replicaId); + if (offsetData != null) { + bld.append(", offsetData=").append(offsetData); + } + if (partitionTimestamps != null) { + bld.append(", partitionTimestamps=").append(partitionTimestamps); + } + bld.append(", isolationLevel=").append(isolationLevel); + bld.append(")"); + return bld.toString(); + } + } + + /** + * This class is only used by ListOffsetRequest v0 which has been deprecated. + */ + @Deprecated + public static final class PartitionData { + public final long timestamp; + public final int maxNumOffsets; + + public PartitionData(long timestamp, int maxNumOffsets) { + this.timestamp = timestamp; + this.maxNumOffsets = maxNumOffsets; + } + + @Override + public String toString() { + StringBuilder bld = new StringBuilder(); + bld.append("{timestamp: ").append(timestamp). + append(", maxNumOffsets: ").append(maxNumOffsets). + append("}"); + return bld.toString(); + } + } + + /** + * Private constructor with a specified version. + */ + @SuppressWarnings("unchecked") + private ListOffsetRequestV0(int replicaId, Map targetTimes, + IsolationLevel isolationLevel, short version) { + super(ApiKeys.LIST_OFFSETS, version); + this.replicaId = replicaId; + this.isolationLevel = isolationLevel; + this.offsetData = version == 0 ? (Map) targetTimes : null; + this.partitionTimestamps = version >= 1 ? (Map) targetTimes : null; + this.duplicatePartitions = Collections.emptySet(); + } + + public ListOffsetRequestV0(Struct struct, short version) { + super(ApiKeys.LIST_OFFSETS, version); + Set duplicatePartitions = new HashSet<>(); + replicaId = struct.getInt(REPLICA_ID_KEY_NAME); + isolationLevel = struct.hasField(ISOLATION_LEVEL_KEY_NAME) + ? IsolationLevel.forId(struct.getByte(ISOLATION_LEVEL_KEY_NAME)) : IsolationLevel.READ_UNCOMMITTED; + offsetData = new HashMap<>(); + partitionTimestamps = new HashMap<>(); + for (Object topicResponseObj : struct.getArray(TOPICS_KEY_NAME)) { + Struct topicResponse = (Struct) topicResponseObj; + String topic = topicResponse.get(TOPIC_NAME); + for (Object partitionResponseObj : topicResponse.getArray(PARTITIONS_KEY_NAME)) { + Struct partitionResponse = (Struct) partitionResponseObj; + int partition = partitionResponse.get(PARTITION_ID); + long timestamp = partitionResponse.getLong(TIMESTAMP_KEY_NAME); + TopicPartition tp = new TopicPartition(topic, partition); + if (partitionResponse.hasField(MAX_NUM_OFFSETS_KEY_NAME)) { + int maxNumOffsets = partitionResponse.getInt(MAX_NUM_OFFSETS_KEY_NAME); + ListOffsetRequestV0.PartitionData partitionData = + new ListOffsetRequestV0.PartitionData(timestamp, maxNumOffsets); + offsetData.put(tp, partitionData); + } else { + if (partitionTimestamps.put(tp, timestamp) != null){ + duplicatePartitions.add(tp); + } + } + } + } + this.duplicatePartitions = duplicatePartitions; + } + + @Override + @SuppressWarnings("deprecation") + public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) { + Map responseData = new HashMap<>(); + + short versionId = version(); + if (versionId == 0) { + for (Map.Entry entry : offsetData.entrySet()) { + ListOffsetResponse.PartitionData partitionResponse = new ListOffsetResponse.PartitionData( + Errors.forException(e), Collections.emptyList()); + responseData.put(entry.getKey(), partitionResponse); + } + } else { + for (Map.Entry entry : partitionTimestamps.entrySet()) { + ListOffsetResponse.PartitionData partitionResponse = new ListOffsetResponse.PartitionData( + Errors.forException(e), -1L, -1L, Optional.empty()); + responseData.put(entry.getKey(), partitionResponse); + } + } + + switch (versionId) { + case 0: + case 1: + case 2: + case 3: + return new ListOffsetResponse(throttleTimeMs, responseData); + default: + throw new IllegalArgumentException( + String.format("Version %d is not valid. Valid versions for %s are 0 to %d", + versionId, this.getClass().getSimpleName(), ApiKeys.LIST_OFFSETS.latestVersion())); + } + } + + public int replicaId() { + return replicaId; + } + + public IsolationLevel isolationLevel() { + return isolationLevel; + } + + @Deprecated + public Map offsetData() { + return offsetData; + } + + public Map partitionTimestamps() { + return partitionTimestamps; + } + + public Set duplicatePartitions() { + return duplicatePartitions; + } + + public static ListOffsetRequestV0 parse(ByteBuffer buffer, short version) { + return new ListOffsetRequestV0(ApiKeys.LIST_OFFSETS.parseRequest(version, buffer), version); + } + + @Override + protected Struct toStruct() { + short version = version(); + Struct struct = new Struct(ApiKeys.LIST_OFFSETS.requestSchema(version)); + + Map targetTimes = partitionTimestamps == null ? offsetData : partitionTimestamps; + Map> topicsData = groupDataByTopic(targetTimes); + + struct.set(REPLICA_ID_KEY_NAME, replicaId); + + if (struct.hasField(ISOLATION_LEVEL_KEY_NAME)){ + struct.set(ISOLATION_LEVEL_KEY_NAME, isolationLevel.id()); + } + List topicArray = new ArrayList<>(); + for (Map.Entry> topicEntry: topicsData.entrySet()) { + Struct topicData = struct.instance(TOPICS_KEY_NAME); + topicData.set(TOPIC_NAME, topicEntry.getKey()); + List partitionArray = new ArrayList<>(); + for (Map.Entry partitionEntry : topicEntry.getValue().entrySet()) { + if (version == 0) { + ListOffsetRequestV0.PartitionData offsetPartitionData = + (ListOffsetRequestV0.PartitionData) partitionEntry.getValue(); + Struct partitionData = topicData.instance(PARTITIONS_KEY_NAME); + partitionData.set(PARTITION_ID, partitionEntry.getKey()); + partitionData.set(TIMESTAMP_KEY_NAME, offsetPartitionData.timestamp); + partitionData.set(MAX_NUM_OFFSETS_KEY_NAME, offsetPartitionData.maxNumOffsets); + partitionArray.add(partitionData); + } else { + Long timestamp = (Long) partitionEntry.getValue(); + Struct partitionData = topicData.instance(PARTITIONS_KEY_NAME); + partitionData.set(PARTITION_ID, partitionEntry.getKey()); + partitionData.set(TIMESTAMP_KEY_NAME, timestamp); + partitionArray.add(partitionData); + } + } + topicData.set(PARTITIONS_KEY_NAME, partitionArray.toArray()); + topicArray.add(topicData); + } + struct.set(TOPICS_KEY_NAME, topicArray.toArray()); + return struct; + } + + public static Map> groupDataByTopic(Map data) { + Map> dataByTopic = new HashMap<>(); + for (Map.Entry entry: data.entrySet()) { + String topic = entry.getKey().topic(); + int partition = entry.getKey().partition(); + Map topicData = dataByTopic.computeIfAbsent(topic, k -> new HashMap<>()); + topicData.put(partition, entry.getValue()); + } + return dataByTopic; + } +} diff --git a/pom.xml b/pom.xml index 3cc13aa289..d3d513c724 100644 --- a/pom.xml +++ b/pom.xml @@ -87,7 +87,6 @@ kafka-payload-processor-shaded kafka-payload-processor-shaded-tests test-listener - kafka-client-2-0-shade @@ -99,12 +98,6 @@ ${kafka.version} - - io.streamnative.pulsar.handlers - kafka-client-2-0-shade - ${project.version} - - ${pulsar.group.id} pulsar-broker From 43ed5403bd69f055d454eb17b1cff3c7e234a104 Mon Sep 17 00:00:00 2001 From: Demogorgon314 Date: Thu, 26 May 2022 18:40:33 +0800 Subject: [PATCH 4/4] Fix codacy check --- .../handlers/kop/KafkaRequestHandler.java | 13 ++++++------- .../common/requests/ListOffsetRequestV0.java | 18 +++++++++--------- 2 files changed, 15 insertions(+), 16 deletions(-) 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 448fabaae5..e08e7df468 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 @@ -1080,13 +1080,7 @@ protected void handleOffsetFetchRequest(KafkaHeaderAndRequest offsetFetch, private CompletableFuture> fetchOffset(String topicName, long timestamp) { CompletableFuture> partitionData = new CompletableFuture<>(); - topicManager.getTopic(topicName).whenComplete((perTopicOpt, t) -> { - if (t != null) { - log.error("Failed while get persistentTopic topic: {} ts: {}. ", - !perTopicOpt.isPresent() ? "null" : perTopicOpt.get().getName(), timestamp, t); - partitionData.complete(Pair.of(Errors.forException(t), null)); - return; - } + topicManager.getTopic(topicName).thenAccept((perTopicOpt) -> { if (!perTopicOpt.isPresent()) { partitionData.complete(Pair.of(Errors.UNKNOWN_TOPIC_OR_PARTITION, null)); return; @@ -1147,6 +1141,11 @@ private CompletableFuture> fetchOffset(String topicName, long } else { fetchOffsetByTimestamp(partitionData, managedLedger, lac, timestamp, perTopic.getName()); } + }).exceptionally(e -> { + Throwable throwable = FutureUtil.unwrapCompletionException(e); + log.error("Failed while get persistentTopic topic: {} ts: {}. ", topicName, timestamp, throwable); + partitionData.complete(Pair.of(Errors.forException(throwable), null)); + return null; }); return partitionData; diff --git a/kafka-impl/src/main/java/org/apache/kafka/common/requests/ListOffsetRequestV0.java b/kafka-impl/src/main/java/org/apache/kafka/common/requests/ListOffsetRequestV0.java index e972a9db7f..f6f52746be 100644 --- a/kafka-impl/src/main/java/org/apache/kafka/common/requests/ListOffsetRequestV0.java +++ b/kafka-impl/src/main/java/org/apache/kafka/common/requests/ListOffsetRequestV0.java @@ -94,11 +94,6 @@ public class ListOffsetRequestV0 extends AbstractRequest { */ private static final Schema LIST_OFFSET_REQUEST_V3 = LIST_OFFSET_REQUEST_V2; - public static Schema[] schemaVersions() { - return new Schema[] {LIST_OFFSET_REQUEST_V0, LIST_OFFSET_REQUEST_V1, LIST_OFFSET_REQUEST_V2, - LIST_OFFSET_REQUEST_V3}; - } - private final int replicaId; private final IsolationLevel isolationLevel; private final Map offsetData; @@ -151,8 +146,8 @@ public ListOffsetRequestV0 build(short version) { if (version == 0) { if (offsetData == null) { if (partitionTimestamps == null) { - throw new RuntimeException("Must set partitionTimestamps or offsetData when creating a v0 " - + "ListOffsetRequest"); + throw new IllegalArgumentException( + "Must set partitionTimestamps or offsetData when creating a v0 ListOffsetRequest"); } else { offsetData = new HashMap<>(); for (Map.Entry entry: partitionTimestamps.entrySet()) { @@ -164,10 +159,10 @@ public ListOffsetRequestV0 build(short version) { } } else { if (offsetData != null) { - throw new RuntimeException("Cannot create a v" + version + " ListOffsetRequest with v0 " + throw new IllegalArgumentException("Cannot create a v" + version + " ListOffsetRequest with v0 " + "PartitionData."); } else if (partitionTimestamps == null) { - throw new RuntimeException("Must set partitionTimestamps when creating a v" + throw new IllegalArgumentException("Must set partitionTimestamps when creating a v" + version + " ListOffsetRequest"); } } @@ -360,6 +355,11 @@ protected Struct toStruct() { return struct; } + public static Schema[] schemaVersions() { + return new Schema[] {LIST_OFFSET_REQUEST_V0, LIST_OFFSET_REQUEST_V1, LIST_OFFSET_REQUEST_V2, + LIST_OFFSET_REQUEST_V3}; + } + public static Map> groupDataByTopic(Map data) { Map> dataByTopic = new HashMap<>(); for (Map.Entry entry: data.entrySet()) {