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..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,6 +155,14 @@ protected KafkaHeaderAndRequest byteBufToRequest(ByteBuf msg, } } + protected ListOffsetRequestV0 byteBufToListOffsetRequestV0(ByteBuf buf) { + checkArgument(buf.readableBytes() > 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 d6836f8c27..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 @@ -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; @@ -1076,17 +1077,10 @@ 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) -> { - 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, List } 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; @@ -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,62 @@ 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"))); + ListOffsetRequestV0 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 -> { + 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(topic, CompletableFuture.completedFuture( + Pair.of(Errors.TOPIC_AUTHORIZATION_FAILED, null))); + completeOne.run(); + return; + } + if (!isAuthorized) { + responseData.put(topic, 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(topic, 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..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 @@ -16,9 +16,12 @@ 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; +import org.apache.kafka.common.requests.ListOffsetRequestV0; import org.apache.kafka.common.requests.OffsetCommitRequest; import org.apache.kafka.common.requests.TxnOffsetCommitRequest; @@ -44,6 +47,14 @@ public static long getOffset(TxnOffsetCommitRequest.CommittedOffset committedOff public static class LegacyUtils { + public static void forEachListOffsetRequest( + ListOffsetRequestV0 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/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..f6f52746be --- /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; + + 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/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 -> {