diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/AdminManager.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/AdminManager.java index f0295fb59c..59f509a116 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/AdminManager.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/AdminManager.java @@ -40,7 +40,6 @@ import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.mledger.Position; import org.apache.bookkeeper.mledger.impl.PositionImpl; -import org.apache.kafka.clients.admin.NewPartitions; import org.apache.kafka.common.Node; import org.apache.kafka.common.config.ConfigResource; import org.apache.kafka.common.errors.InvalidPartitionsException; @@ -49,6 +48,7 @@ import org.apache.kafka.common.errors.UnknownTopicOrPartitionException; import org.apache.kafka.common.protocol.Errors; import org.apache.kafka.common.requests.ApiError; +import org.apache.kafka.common.requests.CreatePartitionsRequest; import org.apache.kafka.common.requests.CreateTopicsRequest; import org.apache.kafka.common.requests.DescribeConfigsResponse; import org.apache.kafka.common.requests.MetadataResponse; @@ -299,9 +299,10 @@ public void truncateTopic(String topicToDelete, } - CompletableFuture> createPartitionsAsync(Map createInfo, - int timeoutMs, - String namespacePrefix) { + CompletableFuture> createPartitionsAsync( + Map createInfo, + int timeoutMs, + String namespacePrefix) { final Map> futureMap = new ConcurrentHashMap<>(); final AtomicInteger numTopics = new AtomicInteger(createInfo.size()); final CompletableFuture> resultFuture = new CompletableFuture<>(); @@ -336,12 +337,12 @@ CompletableFuture> createPartitionsAsync(Map 0); + ByteBuffer nio = buf.nioBuffer(); + RequestHeader header = RequestHeader.parse(nio); + short apiVersion = header.apiVersion(); + return ListOffsetRequestV0.parse(nio, apiVersion); + } + 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 729fbe29b3..291a8dd2cc 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 @@ -92,7 +92,6 @@ import org.apache.commons.collections4.ListUtils; import org.apache.commons.lang3.NotImplementedException; import org.apache.commons.lang3.tuple.Pair; -import org.apache.kafka.clients.admin.NewPartitions; import org.apache.kafka.common.Node; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.acl.AclOperation; @@ -143,6 +142,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; @@ -1285,7 +1285,7 @@ private void handleListOffsetRequestV1AndAbove(KafkaHeaderAndRequest listOffset, completeOne.run(); return; } - responseData.put(topic, fetchOffset(fullPartitionName, times)); + responseData.put(topic, fetchOffset(fullPartitionName, times.timestamp)); completeOne.run(); } ); @@ -1298,7 +1298,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) { - ListOffsetRequest request = (ListOffsetRequest) listOffset.getRequest(); + ListOffsetRequestV0 request = (ListOffsetRequestV0) listOffset.getRequest(); Map>> responseData = Maps.newConcurrentMap(); @@ -1359,7 +1359,8 @@ private void handleListOffsetRequestV0(KafkaHeaderAndRequest listOffset, @Override protected void handleListOffsetRequest(KafkaHeaderAndRequest listOffset, CompletableFuture resultFuture) { - checkArgument(listOffset.getRequest() instanceof ListOffsetRequest); + checkArgument(listOffset.getRequest() instanceof ListOffsetRequest + || listOffset.getRequest() instanceof ListOffsetRequestV0); // the only difference between v0 and v1 is the `max_num_offsets => INT32` // v0 is required because it is used by librdkafka if (listOffset.getHeader().apiVersion() == 0) { @@ -2451,7 +2452,7 @@ protected void handleCreatePartitions(KafkaHeaderAndRequest createPartitions, CreatePartitionsRequest request = (CreatePartitionsRequest) createPartitions.getRequest(); final Map result = Maps.newConcurrentMap(); - final Map validTopics = Maps.newHashMap(); + final Map validTopics = Maps.newHashMap(); final Set duplicateTopics = request.duplicates(); KafkaRequestUtils.forEachCreatePartitionsRequest(request, (topic, newPartition) -> { @@ -2472,7 +2473,7 @@ protected void handleCreatePartitions(KafkaHeaderAndRequest createPartitions, String namespacePrefix = currentNamespacePrefix(); final AtomicInteger validTopicsCount = new AtomicInteger(validTopics.size()); - final Map authorizedTopics = Maps.newConcurrentMap(); + final Map authorizedTopics = Maps.newConcurrentMap(); Runnable createPartitionsAsync = () -> { if (authorizedTopics.isEmpty()) { resultFuture.complete(KafkaResponseUtils.newCreatePartitions(result)); @@ -2486,7 +2487,7 @@ protected void handleCreatePartitions(KafkaHeaderAndRequest createPartitions, }); }; - BiConsumer completeOneTopic = (topic, newPartitions) -> { + BiConsumer completeOneTopic = (topic, newPartitions) -> { authorizedTopics.put(topic, newPartitions); if (validTopicsCount.decrementAndGet() == 0) { createPartitionsAsync.run(); diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/coordinator/transaction/TransactionMarkerChannelHandler.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/coordinator/transaction/TransactionMarkerChannelHandler.java index 4596f1f2e1..978cb000bd 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/coordinator/transaction/TransactionMarkerChannelHandler.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/coordinator/transaction/TransactionMarkerChannelHandler.java @@ -27,6 +27,7 @@ import java.util.List; import java.util.concurrent.CompletableFuture; import java.util.function.Consumer; +import javax.security.sasl.SaslException; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.util.collections.ConcurrentLongHashMap; import org.apache.kafka.common.protocol.ApiKeys; @@ -221,7 +222,7 @@ private CompletableFuture authenticateInternal(ChannelHan saslAuthBytes = usernamePassword.getBytes(UTF_8); break; case OAuthBearerLoginModule.OAUTHBEARER_MECHANISM: - saslAuthBytes = new OAuthBearerClientInitialResponse(commandData).toBytes(); + saslAuthBytes = new OAuthBearerClientInitialResponse(commandData, null, null).toBytes(); break; default: log.error("No corresponding mechanism to {}", authentication.getClass().getName()); @@ -252,7 +253,7 @@ private CompletableFuture authenticateInternal(ChannelHan result.completeExceptionally(saslResponse.error().exception()); } }); - } catch (PulsarClientException ex) { + } catch (PulsarClientException | SaslException ex) { log.error("Transaction marker channel handler authentication failed.", ex); result.completeExceptionally(ex); } 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 266b58150f..ea76880227 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 @@ -18,37 +18,38 @@ import java.util.function.BiConsumer; import java.util.function.Consumer; import java.util.function.Function; -import org.apache.kafka.clients.admin.NewPartitions; 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; public class KafkaRequestUtils { - public static void forEachCreatePartitionsRequest(CreatePartitionsRequest request, - BiConsumer consumer) { + public static void forEachCreatePartitionsRequest( + CreatePartitionsRequest request, + BiConsumer consumer) { request.newPartitions().forEach(consumer); } public static void forEachListOffsetRequest(ListOffsetRequest request, - BiConsumer consumer) { + BiConsumer consumer) { request.partitionTimestamps().forEach(consumer); } public static String getMetadata(TxnOffsetCommitRequest.CommittedOffset committedOffset) { - return Optional.ofNullable(committedOffset.metadata()).orElse(OffsetAndMetadata.NoMetadata); + return Optional.ofNullable(committedOffset.metadata).orElse(OffsetAndMetadata.NoMetadata); } public static long getOffset(TxnOffsetCommitRequest.CommittedOffset committedOffset) { - return committedOffset.offset(); + return committedOffset.offset; } public static class LegacyUtils { public static void forEachListOffsetRequest( - ListOffsetRequest request, + 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/io/streamnative/pulsar/handlers/kop/utils/KafkaResponseUtils.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/KafkaResponseUtils.java index 87a1cfe8fc..0383db7ada 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/KafkaResponseUtils.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/KafkaResponseUtils.java @@ -147,7 +147,7 @@ public static ListGroupsResponse newListGroups(Errors errors, public static ListOffsetResponse newListOffset( Map> partitionToOffset, + Pair> partitionToOffset, boolean legacy) { if (legacy) { return new ListOffsetResponse(CoreUtils.mapValue(partitionToOffset, @@ -155,13 +155,16 @@ public static ListOffsetResponse newListOffset( pair.getLeft(), Optional.ofNullable(pair.getRight()).map(Collections::singletonList) .orElse(Collections.emptyList())) - )); + )); } else { return new ListOffsetResponse(CoreUtils.mapValue(partitionToOffset, pair -> new ListOffsetResponse.PartitionData( pair.getLeft(), // error 0L, // timestamp - Optional.ofNullable(pair.getRight()).orElse(0L) // offset + Optional.ofNullable( + pair.getRight() != null ? pair.getRight().intValue() : null) + .orElse(0) // offset + , Optional.empty() ) )); } @@ -179,6 +182,7 @@ public static MetadataResponse.PartitionMetadata newMetadataPartition(int partit return new MetadataResponse.PartitionMetadata(Errors.NONE, partition, node, // leader + Optional.empty(), // leaderEpoch is unknown in Pulsar Collections.singletonList(node), // replicas Collections.singletonList(node), // isr Collections.emptyList() // offline replicas @@ -190,6 +194,7 @@ public static MetadataResponse.PartitionMetadata newMetadataPartition(Errors err return new MetadataResponse.PartitionMetadata(errors, partition, Node.noNode(), // leader + Optional.empty(), // leaderEpoch is unknown in Pulsar Collections.singletonList(Node.noNode()), // replicas Collections.singletonList(Node.noNode()), // isr Collections.emptyList() // offline replicas @@ -203,12 +208,14 @@ public static OffsetCommitResponse newOffsetCommit(Map r public static OffsetFetchResponse.PartitionData newOffsetFetchPartition(long offset, String metadata) { return new OffsetFetchResponse.PartitionData(offset, + Optional.empty(), // leaderEpoch is unknown in Pulsar metadata, Errors.NONE); } public static OffsetFetchResponse.PartitionData newOffsetFetchPartition() { return new OffsetFetchResponse.PartitionData(OffsetFetchResponse.INVALID_OFFSET, + Optional.empty(), // leaderEpoch is unknown in Pulsar "", // metadata Errors.NONE ); 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..70e8553835 --- /dev/null +++ b/kafka-impl/src/main/java/org/apache/kafka/common/requests/ListOffsetRequestV0.java @@ -0,0 +1,374 @@ +/** + * 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; + + 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 IllegalArgumentException( + "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 IllegalArgumentException("Cannot create a v" + version + " ListOffsetRequest with v0 " + + "PartitionData."); + } else if (partitionTimestamps == null) { + throw new IllegalArgumentException("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 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()) { + 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 dd8150852f..2b0e1d46b1 100644 --- a/pom.xml +++ b/pom.xml @@ -32,7 +32,7 @@ true 2.14.2 - 2.0.0 + 2.1.1 1.18.24 4.11.0 io.streamnative diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaApisTest.java b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaApisTest.java index 385c9e2f06..8270c0265f 100644 --- a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaApisTest.java +++ b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaApisTest.java @@ -1046,7 +1046,7 @@ public void testFetchMinBytesSingleConsumer() throws Exception { @Cleanup final KafkaHeaderAndRequest request = buildRequest(FetchRequest.Builder.forConsumer(maxWaitMs, minBytes, Collections.singletonMap(topicPartition, new FetchRequest.PartitionData( - 0L, -1L, 1024 * 1024 + 0L, -1L, 1024 * 1024, Optional.empty() )))); final CompletableFuture future = new CompletableFuture<>(); final long startTime = System.currentTimeMillis(); diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaCommonTestUtils.java b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaCommonTestUtils.java index 384b54f025..60eeee700a 100644 --- a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaCommonTestUtils.java +++ b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaCommonTestUtils.java @@ -16,19 +16,21 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.stream.Collectors; -import org.apache.kafka.clients.admin.NewPartitions; import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.common.requests.CreatePartitionsRequest; import org.apache.kafka.common.requests.FetchRequest; +import org.apache.kafka.common.requests.ListOffsetRequest; import org.apache.kafka.common.requests.OffsetCommitRequest; import org.apache.kafka.common.requests.TxnOffsetCommitRequest; public class KafkaCommonTestUtils { - public static Map newListOffsetTargetTimes( + public static Map newListOffsetTargetTimes( TopicPartition topicPartition, long timestamp) { - return Collections.singletonMap(topicPartition, timestamp); + return Collections.singletonMap(topicPartition, new ListOffsetRequest.PartitionData(timestamp, 100)); } public static FetchRequest.PartitionData newFetchRequestPartitionData(long fetchOffset, @@ -36,7 +38,8 @@ public static FetchRequest.PartitionData newFetchRequestPartitionData(long fetch int maxBytes) { return new FetchRequest.PartitionData(fetchOffset, logStartOffset, - maxBytes + maxBytes, + Optional.empty() ); } @@ -44,23 +47,27 @@ public static TxnOffsetCommitRequest.CommittedOffset newTxnOffsetCommitRequestCo long offset, String metadata) { return new TxnOffsetCommitRequest.CommittedOffset(offset, - metadata + metadata, + Optional.empty() ); } public static OffsetCommitRequest.PartitionData newOffsetCommitRequestPartitionData(long offset, String metadata) { return new OffsetCommitRequest.PartitionData(offset, + Optional.empty(), metadata ); } - public static Map newPartitionsMap(List topics, int totalCount) { - return topics.stream().collect(Collectors.toMap(topic -> topic, __ -> NewPartitions.increaseTo(totalCount))); + public static Map newPartitionsMap( + List topics, int totalCount) { + return topics.stream().collect(Collectors.toMap(topic -> topic, + __ -> new CreatePartitionsRequest.PartitionDetails(totalCount))); } - public static Map newPartitionsMap(String topic, int totalCount) { + public static Map newPartitionsMap(String topic, int totalCount) { return newPartitionsMap(Collections.singletonList(topic), totalCount); } } diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/SaslOAuthBearerTestBase.java b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/SaslOAuthBearerTestBase.java index 652e5e2507..2e9f268a86 100644 --- a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/SaslOAuthBearerTestBase.java +++ b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/SaslOAuthBearerTestBase.java @@ -88,7 +88,8 @@ protected void testProduceWithoutAuth() throws Exception { fail("should have failed"); } catch (ExecutionException e) { assertTrue(e.getCause() instanceof TimeoutException); - assertTrue(e.getMessage().contains("Failed to update metadata")); + assertTrue(e.getMessage().contains("Topic " + topic + + " not present in metadata after 3000 ms.")); } } } diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/SaslPlainTestBase.java b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/SaslPlainTestBase.java index 3321dd33e2..a3dbc4368a 100644 --- a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/SaslPlainTestBase.java +++ b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/SaslPlainTestBase.java @@ -221,7 +221,8 @@ void clientWithoutAuth() throws Exception { fail("should have failed"); } catch (ExecutionException e) { assertTrue(e.getCause() instanceof TimeoutException); - assertTrue(e.getMessage().contains("Failed to update metadata")); + assertTrue(e.getMessage().contains("Topic " + TOPIC + + " not present in metadata after " + metadataTimeoutMs + " ms.")); } } 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 12e1f274c2..8f0621066a 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 @@ -354,8 +354,9 @@ void clientWithoutAuth() throws Exception { producer.newContextBuilder(KAFKA_TOPIC, "hello").build().sendAsync().get(); fail("should have failed"); } catch (Exception e) { - if (version == KafkaVersion.KAFKA_2_8_0 - || version == KafkaVersion.KAFKA_3_0_0) { + if (version == KafkaVersion.DEFAULT + || version == KafkaVersion.KAFKA_2_8_0 + || version == KafkaVersion.KAFKA_3_0_0) { assertTrue(e.getMessage().contains("Topic " + KAFKA_TOPIC + " not present in metadata after " + metadataTimeoutMs + " ms.")); } else {