chia7712 commented on code in PR #16873:
URL: https://github.com/apache/kafka/pull/16873#discussion_r1732061473
##########
core/src/main/scala/kafka/server/KafkaApis.scala:
##########
@@ -1103,35 +1103,41 @@ class KafkaApis(val requestChannel: RequestChannel,
val responseTopics = authorizedRequestInfo.map { topic =>
val responsePartitions = topic.partitions.asScala.map { partition =>
- val topicPartition = new TopicPartition(topic.name,
partition.partitionIndex)
-
- try {
- val offsets = replicaManager.legacyFetchOffsetsForTimestamp(
- topicPartition = topicPartition,
- timestamp = partition.timestamp,
- maxNumOffsets = partition.maxNumOffsets,
- isFromConsumer = offsetRequest.replicaId ==
ListOffsetsRequest.CONSUMER_REPLICA_ID,
- fetchOnlyFromLeader = offsetRequest.replicaId !=
ListOffsetsRequest.DEBUGGING_REPLICA_ID)
+ if (partition.timestamp() < ListOffsetsRequest.EARLIEST_TIMESTAMP) {
new ListOffsetsPartitionResponse()
.setPartitionIndex(partition.partitionIndex)
- .setErrorCode(Errors.NONE.code)
- .setOldStyleOffsets(offsets.map(JLong.valueOf).asJava)
- } catch {
- // NOTE: UnknownTopicOrPartitionException and
NotLeaderOrFollowerException are special cases since these error messages
- // are typically transient and there is no value in logging the
entire stack trace for the same
- case e @ (_ : UnknownTopicOrPartitionException |
- _ : NotLeaderOrFollowerException |
- _ : KafkaStorageException) =>
- debug("Offset request with correlation id %d from client %s on
partition %s failed due to %s".format(
- correlationId, clientId, topicPartition, e.getMessage))
- new ListOffsetsPartitionResponse()
- .setPartitionIndex(partition.partitionIndex)
- .setErrorCode(Errors.forException(e).code)
- case e: Throwable =>
- error("Error while responding to offset request", e)
+ .setErrorCode(Errors.UNSUPPORTED_VERSION.code)
+ } else {
+ val topicPartition = new TopicPartition(topic.name,
partition.partitionIndex)
+
+ try {
+ val offsets = replicaManager.legacyFetchOffsetsForTimestamp(
Review Comment:
> This still doesn't support enabling remote storage dynamically, but is
probably good enough in 3.9.
enabling remote storage is another issue, and I agree to your approach is
good and simple in 3.9
> what happens if the MV gets downgraded?
that is a good question. In the request-level, both version=14 of `FETCH`
and version=8 of `LIST_OFFSET` are supported by 3.5-IV2. Hence, it should be
fine even though MV gets safe-downgraded to 3.5-IV2.
> I am not sure if tier storage supports downgrading MV
I guess that will be a issue in the future if tier storage requires more and
more new requests to address more features.
IMHO we can adopt @junrao's approach - checking the cluster MV (3.6) when
creating tier storage. The side effect is that we will "allow" safe-downgrade
to 3.5-IV2 even though tiered storage is enabled. That is a bit weird since we
declare tier storage is early access feature available as of 3.6.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
