Skip to content
This repository was archived by the owner on Jan 24, 2024. It is now read-only.

Restore the support for Kafka clients 0.9 and 0.10 #1307

Merged
BewareMyPower merged 4 commits into from
May 30, 2022
Merged

Restore the support for Kafka clients 0.9 and 0.10 #1307

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
16f1e5e
Restore the support for Kafka clients 0.9 and 0.10
Demogorgon314 May 26, 2022
8190682
Remove unused code
Demogorgon314 May 26, 2022
26127c5
Copy ListOffsetRequest to KoP
Demogorgon314 May 26, 2022
43ed540
Fix codacy check
Demogorgon314 May 26, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaCommandDecoder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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)) {
Expand Down
79 changes: 65 additions & 14 deletions kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -1076,17 +1077,10 @@ protected void handleOffsetFetchRequest(KafkaHeaderAndRequest offsetFetch,
});
}

private CompletableFuture<Pair<Errors, Long>> fetchOffset(String topicName, ListOffsetRequest.PartitionData pd) {
Long timestamp = pd.timestamp;
private CompletableFuture<Pair<Errors, Long>> fetchOffset(String topicName, long timestamp) {
CompletableFuture<Pair<Errors, Long>> 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;
Expand Down Expand Up @@ -1147,6 +1141,11 @@ private CompletableFuture<Pair<Errors, Long>> 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;
Expand Down Expand Up @@ -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();
}
);
Expand All @@ -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<AbstractResponse> 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<TopicPartition, CompletableFuture<Pair<Errors, Long>>> 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<Pair<Errors, Long>> 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
Expand Down
11 changes: 11 additions & 0 deletions kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/utils/KafkaRequestUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -44,6 +47,14 @@ public static long getOffset(TxnOffsetCommitRequest.CommittedOffset committedOff

public static class LegacyUtils {

public static void forEachListOffsetRequest(
ListOffsetRequestV0 request,
Function<TopicPartition, Function<Long, Consumer<Integer>>> 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();
Expand Down
Loading