You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "TaiJuWu (via GitHub)" <gi...@apache.org> on 2025/09/18 06:24:23 UTC
[PR] (WIP) KAFKA-19706: ListOffsetsHandler should kick out the unsupported timestyamp type when receiving a UnsupportedVersionException [kafka]
TaiJuWu opened a new pull request, #20553:
URL: https://github.com/apache/kafka/pull/20553
There are two issues need to address:
1. UnSupportedVersion is count to tries.
2. MockClient disconnect after downgrade two timestamps
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19706: ListOffsetsHandler should kick out the unsupported timestamp [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #20553:
URL: https://github.com/apache/kafka/pull/20553#issuecomment-3336633379
A label of 'needs-attention' was automatically added to this PR in order to raise the
attention of the committers. Once this issue has been triaged, the `triage` label
should be removed to prevent this automation from happening again.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19706: ListOffsetsHandler should kick out the unsupported timestamp [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #20553: KAFKA-19706: ListOffsetsHandler should kick out the unsupported timestamp
URL: https://github.com/apache/kafka/pull/20553
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19706: ListOffsetsHandler should kick out the unsupported timestamp [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #20553:
URL: https://github.com/apache/kafka/pull/20553#issuecomment-3713005285
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19706: ListOffsetsHandler should kick out the unsupported timestamp [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #20553:
URL: https://github.com/apache/kafka/pull/20553#issuecomment-3850984593
This PR has been closed since it has not had any activity in 120 days. If you feel like this
was a mistake, or you would like to continue working on it, please feel free to re-open the
PR and ask for a review.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19706: ListOffsetsHandler should kick out the unsupported timestamp [kafka]
Posted by "zheguang (via GitHub)" <gi...@apache.org>.
zheguang commented on code in PR #20553:
URL: https://github.com/apache/kafka/pull/20553#discussion_r2409002908
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -202,23 +207,61 @@ private void handlePartitionError(
public Map<TopicPartition, Throwable> handleUnsupportedVersionException(
int brokerId, UnsupportedVersionException exception, Set<TopicPartition> keys
) {
- log.warn("Broker {} does not support MAX_TIMESTAMP offset specs", brokerId);
- Map<TopicPartition, Throwable> maxTimestampPartitions = new HashMap<>();
+ log.warn("Broker {} does not support {} offset specs", brokerId, timestampToString(currentUnsupportedVersion));
+ Map<TopicPartition, Throwable> supportedTimestampPartitions = new HashMap<>();
+
for (TopicPartition topicPartition : keys) {
Long offsetTimestamp = offsetTimestampsByPartition.get(topicPartition);
- if (offsetTimestamp == ListOffsetsRequest.MAX_TIMESTAMP) {
- maxTimestampPartitions.put(topicPartition, exception);
+ if (offsetTimestamp != null && offsetTimestamp == currentUnsupportedVersion && !accUnsupportedTimestampPartition.containsKey(topicPartition)) {
+ supportedTimestampPartitions.put(topicPartition, exception);
+ accUnsupportedTimestampPartition.put(topicPartition, exception);
}
}
- // If there are no partitions with MAX_TIMESTAMP specs the UnsupportedVersionException cannot be handled
+
+ // If there are no partitions with support specs the UnsupportedVersionException cannot be handled
// and all partitions should be failed here.
- // Otherwise, just the partitions with MAX_TIMESTAMP specs should be failed here and the fulfillment stage
- // will later be retried for the potentially empty set of partitions with non-MAX_TIMESTAMP specs.
- if (maxTimestampPartitions.isEmpty()) {
+ // Otherwise, just the partitions with support specs should be failed here and the fulfillment stage
+ // will later be retried for the potentially empty set of partitions with non-support specs.
+ if (unsupportedVersionRetry == maxUnsupportedVersionRetry && accUnsupportedTimestampPartition.isEmpty()) {
return keys.stream().collect(Collectors.toMap(k -> k, k -> exception));
} else {
- return maxTimestampPartitions;
+ return supportedTimestampPartitions;
+ }
+ }
+
+ public void downgradeOffsetTimestampVersion() {
+ if (unsupportedVersionRetry >= maxUnsupportedVersionRetry) {
+ throw new IllegalStateException("Index should not larger than " + maxUnsupportedVersionRetry);
}
+
+ unsupportedVersionRetry += 1;
+ currentUnsupportedVersion = ListOffsetsRequest.LEAST_TO_OLDEST_TIMESTAMPS.get(unsupportedVersionRetry);
+ }
+
+ public void reset() {
+ currentUnsupportedVersion = ListOffsetsRequest.LEAST_TO_OLDEST_TIMESTAMPS.get(0);
+ unsupportedVersionRetry = 0;
+ accUnsupportedTimestampPartition.clear();
+ }
+
+ public boolean isOldestTimstamp() {
Review Comment:
Typo: `isOldestTimstamp` ~> `isOldestTimestamp`
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -202,23 +207,61 @@ private void handlePartitionError(
public Map<TopicPartition, Throwable> handleUnsupportedVersionException(
int brokerId, UnsupportedVersionException exception, Set<TopicPartition> keys
) {
- log.warn("Broker {} does not support MAX_TIMESTAMP offset specs", brokerId);
- Map<TopicPartition, Throwable> maxTimestampPartitions = new HashMap<>();
+ log.warn("Broker {} does not support {} offset specs", brokerId, timestampToString(currentUnsupportedVersion));
+ Map<TopicPartition, Throwable> supportedTimestampPartitions = new HashMap<>();
+
for (TopicPartition topicPartition : keys) {
Long offsetTimestamp = offsetTimestampsByPartition.get(topicPartition);
- if (offsetTimestamp == ListOffsetsRequest.MAX_TIMESTAMP) {
- maxTimestampPartitions.put(topicPartition, exception);
+ if (offsetTimestamp != null && offsetTimestamp == currentUnsupportedVersion && !accUnsupportedTimestampPartition.containsKey(topicPartition)) {
+ supportedTimestampPartitions.put(topicPartition, exception);
+ accUnsupportedTimestampPartition.put(topicPartition, exception);
}
}
- // If there are no partitions with MAX_TIMESTAMP specs the UnsupportedVersionException cannot be handled
+
+ // If there are no partitions with support specs the UnsupportedVersionException cannot be handled
// and all partitions should be failed here.
- // Otherwise, just the partitions with MAX_TIMESTAMP specs should be failed here and the fulfillment stage
- // will later be retried for the potentially empty set of partitions with non-MAX_TIMESTAMP specs.
- if (maxTimestampPartitions.isEmpty()) {
+ // Otherwise, just the partitions with support specs should be failed here and the fulfillment stage
+ // will later be retried for the potentially empty set of partitions with non-support specs.
+ if (unsupportedVersionRetry == maxUnsupportedVersionRetry && accUnsupportedTimestampPartition.isEmpty()) {
return keys.stream().collect(Collectors.toMap(k -> k, k -> exception));
} else {
- return maxTimestampPartitions;
+ return supportedTimestampPartitions;
+ }
+ }
+
+ public void downgradeOffsetTimestampVersion() {
+ if (unsupportedVersionRetry >= maxUnsupportedVersionRetry) {
+ throw new IllegalStateException("Index should not larger than " + maxUnsupportedVersionRetry);
}
+
+ unsupportedVersionRetry += 1;
+ currentUnsupportedVersion = ListOffsetsRequest.LEAST_TO_OLDEST_TIMESTAMPS.get(unsupportedVersionRetry);
+ }
+
+ public void reset() {
+ currentUnsupportedVersion = ListOffsetsRequest.LEAST_TO_OLDEST_TIMESTAMPS.get(0);
+ unsupportedVersionRetry = 0;
+ accUnsupportedTimestampPartition.clear();
+ }
+
+ public boolean isOldestTimstamp() {
+ return currentUnsupportedVersion == ListOffsetsRequest.LEAST_TO_OLDEST_TIMESTAMPS.get(
Review Comment:
This `get()` could simplify to `getLast()`.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/AdminApiDriver.java:
##########
@@ -284,6 +284,17 @@ public void onFailure(
(UnsupportedVersionException) t,
spec.keys);
completeExceptionally(unrecoverableFailures);
+ if (handler instanceof ListOffsetsHandler) {
+ // We dont need to do other operations because completeExceptionally
+ // help us to remove lookup and fulfillmentMap, and the downgraded
+ // offsetStamp still in fulfillmentMap so it will generate next request.
+ ListOffsetsHandler listOffsetsHandler = (ListOffsetsHandler) handler;
+ if (!listOffsetsHandler.isOldestTimstamp()) {
Review Comment:
Typo: isOldestTimstamp ~> isOldestTimestamp
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -202,23 +207,61 @@ private void handlePartitionError(
public Map<TopicPartition, Throwable> handleUnsupportedVersionException(
int brokerId, UnsupportedVersionException exception, Set<TopicPartition> keys
) {
- log.warn("Broker {} does not support MAX_TIMESTAMP offset specs", brokerId);
- Map<TopicPartition, Throwable> maxTimestampPartitions = new HashMap<>();
+ log.warn("Broker {} does not support {} offset specs", brokerId, timestampToString(currentUnsupportedVersion));
+ Map<TopicPartition, Throwable> supportedTimestampPartitions = new HashMap<>();
+
for (TopicPartition topicPartition : keys) {
Long offsetTimestamp = offsetTimestampsByPartition.get(topicPartition);
- if (offsetTimestamp == ListOffsetsRequest.MAX_TIMESTAMP) {
- maxTimestampPartitions.put(topicPartition, exception);
+ if (offsetTimestamp != null && offsetTimestamp == currentUnsupportedVersion && !accUnsupportedTimestampPartition.containsKey(topicPartition)) {
+ supportedTimestampPartitions.put(topicPartition, exception);
+ accUnsupportedTimestampPartition.put(topicPartition, exception);
}
}
- // If there are no partitions with MAX_TIMESTAMP specs the UnsupportedVersionException cannot be handled
+
+ // If there are no partitions with support specs the UnsupportedVersionException cannot be handled
// and all partitions should be failed here.
- // Otherwise, just the partitions with MAX_TIMESTAMP specs should be failed here and the fulfillment stage
- // will later be retried for the potentially empty set of partitions with non-MAX_TIMESTAMP specs.
- if (maxTimestampPartitions.isEmpty()) {
+ // Otherwise, just the partitions with support specs should be failed here and the fulfillment stage
+ // will later be retried for the potentially empty set of partitions with non-support specs.
+ if (unsupportedVersionRetry == maxUnsupportedVersionRetry && accUnsupportedTimestampPartition.isEmpty()) {
return keys.stream().collect(Collectors.toMap(k -> k, k -> exception));
} else {
- return maxTimestampPartitions;
+ return supportedTimestampPartitions;
+ }
+ }
+
+ public void downgradeOffsetTimestampVersion() {
+ if (unsupportedVersionRetry >= maxUnsupportedVersionRetry) {
+ throw new IllegalStateException("Index should not larger than " + maxUnsupportedVersionRetry);
}
+
+ unsupportedVersionRetry += 1;
+ currentUnsupportedVersion = ListOffsetsRequest.LEAST_TO_OLDEST_TIMESTAMPS.get(unsupportedVersionRetry);
+ }
+
+ public void reset() {
+ currentUnsupportedVersion = ListOffsetsRequest.LEAST_TO_OLDEST_TIMESTAMPS.get(0);
+ unsupportedVersionRetry = 0;
+ accUnsupportedTimestampPartition.clear();
+ }
+
+ public boolean isOldestTimstamp() {
+ return currentUnsupportedVersion == ListOffsetsRequest.LEAST_TO_OLDEST_TIMESTAMPS.get(
+ ListOffsetsRequest.LEAST_TO_OLDEST_TIMESTAMPS.size() - 1);
+ }
+
+ // Visible for test
+ public long currentUnsupportedVersion() {
+ return currentUnsupportedVersion;
+ }
+
+ // Visible for test
+ public int unsupportedVersionRetry() {
Review Comment:
Same as above, scope could be package-private.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -202,23 +207,61 @@ private void handlePartitionError(
public Map<TopicPartition, Throwable> handleUnsupportedVersionException(
int brokerId, UnsupportedVersionException exception, Set<TopicPartition> keys
) {
- log.warn("Broker {} does not support MAX_TIMESTAMP offset specs", brokerId);
- Map<TopicPartition, Throwable> maxTimestampPartitions = new HashMap<>();
+ log.warn("Broker {} does not support {} offset specs", brokerId, timestampToString(currentUnsupportedVersion));
+ Map<TopicPartition, Throwable> supportedTimestampPartitions = new HashMap<>();
+
for (TopicPartition topicPartition : keys) {
Long offsetTimestamp = offsetTimestampsByPartition.get(topicPartition);
- if (offsetTimestamp == ListOffsetsRequest.MAX_TIMESTAMP) {
- maxTimestampPartitions.put(topicPartition, exception);
+ if (offsetTimestamp != null && offsetTimestamp == currentUnsupportedVersion && !accUnsupportedTimestampPartition.containsKey(topicPartition)) {
+ supportedTimestampPartitions.put(topicPartition, exception);
+ accUnsupportedTimestampPartition.put(topicPartition, exception);
Review Comment:
It's quite confusing that both "supported" and "unsupported" partitions are accumulated here. Could the logic be clarified?
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -202,23 +207,61 @@ private void handlePartitionError(
public Map<TopicPartition, Throwable> handleUnsupportedVersionException(
int brokerId, UnsupportedVersionException exception, Set<TopicPartition> keys
) {
- log.warn("Broker {} does not support MAX_TIMESTAMP offset specs", brokerId);
- Map<TopicPartition, Throwable> maxTimestampPartitions = new HashMap<>();
+ log.warn("Broker {} does not support {} offset specs", brokerId, timestampToString(currentUnsupportedVersion));
+ Map<TopicPartition, Throwable> supportedTimestampPartitions = new HashMap<>();
+
for (TopicPartition topicPartition : keys) {
Long offsetTimestamp = offsetTimestampsByPartition.get(topicPartition);
- if (offsetTimestamp == ListOffsetsRequest.MAX_TIMESTAMP) {
- maxTimestampPartitions.put(topicPartition, exception);
+ if (offsetTimestamp != null && offsetTimestamp == currentUnsupportedVersion && !accUnsupportedTimestampPartition.containsKey(topicPartition)) {
+ supportedTimestampPartitions.put(topicPartition, exception);
+ accUnsupportedTimestampPartition.put(topicPartition, exception);
}
}
- // If there are no partitions with MAX_TIMESTAMP specs the UnsupportedVersionException cannot be handled
+
+ // If there are no partitions with support specs the UnsupportedVersionException cannot be handled
// and all partitions should be failed here.
- // Otherwise, just the partitions with MAX_TIMESTAMP specs should be failed here and the fulfillment stage
- // will later be retried for the potentially empty set of partitions with non-MAX_TIMESTAMP specs.
- if (maxTimestampPartitions.isEmpty()) {
+ // Otherwise, just the partitions with support specs should be failed here and the fulfillment stage
+ // will later be retried for the potentially empty set of partitions with non-support specs.
+ if (unsupportedVersionRetry == maxUnsupportedVersionRetry && accUnsupportedTimestampPartition.isEmpty()) {
return keys.stream().collect(Collectors.toMap(k -> k, k -> exception));
} else {
- return maxTimestampPartitions;
+ return supportedTimestampPartitions;
+ }
+ }
+
+ public void downgradeOffsetTimestampVersion() {
+ if (unsupportedVersionRetry >= maxUnsupportedVersionRetry) {
+ throw new IllegalStateException("Index should not larger than " + maxUnsupportedVersionRetry);
}
+
+ unsupportedVersionRetry += 1;
+ currentUnsupportedVersion = ListOffsetsRequest.LEAST_TO_OLDEST_TIMESTAMPS.get(unsupportedVersionRetry);
+ }
+
+ public void reset() {
+ currentUnsupportedVersion = ListOffsetsRequest.LEAST_TO_OLDEST_TIMESTAMPS.get(0);
+ unsupportedVersionRetry = 0;
+ accUnsupportedTimestampPartition.clear();
+ }
+
+ public boolean isOldestTimstamp() {
+ return currentUnsupportedVersion == ListOffsetsRequest.LEAST_TO_OLDEST_TIMESTAMPS.get(
+ ListOffsetsRequest.LEAST_TO_OLDEST_TIMESTAMPS.size() - 1);
+ }
+
+ // Visible for test
+ public long currentUnsupportedVersion() {
Review Comment:
The scope could be package-private, as only the `ListOffsetsHandlerTest` calls it.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -202,23 +207,61 @@ private void handlePartitionError(
public Map<TopicPartition, Throwable> handleUnsupportedVersionException(
int brokerId, UnsupportedVersionException exception, Set<TopicPartition> keys
) {
- log.warn("Broker {} does not support MAX_TIMESTAMP offset specs", brokerId);
- Map<TopicPartition, Throwable> maxTimestampPartitions = new HashMap<>();
+ log.warn("Broker {} does not support {} offset specs", brokerId, timestampToString(currentUnsupportedVersion));
+ Map<TopicPartition, Throwable> supportedTimestampPartitions = new HashMap<>();
+
for (TopicPartition topicPartition : keys) {
Long offsetTimestamp = offsetTimestampsByPartition.get(topicPartition);
- if (offsetTimestamp == ListOffsetsRequest.MAX_TIMESTAMP) {
- maxTimestampPartitions.put(topicPartition, exception);
+ if (offsetTimestamp != null && offsetTimestamp == currentUnsupportedVersion && !accUnsupportedTimestampPartition.containsKey(topicPartition)) {
+ supportedTimestampPartitions.put(topicPartition, exception);
+ accUnsupportedTimestampPartition.put(topicPartition, exception);
}
}
- // If there are no partitions with MAX_TIMESTAMP specs the UnsupportedVersionException cannot be handled
+
+ // If there are no partitions with support specs the UnsupportedVersionException cannot be handled
// and all partitions should be failed here.
- // Otherwise, just the partitions with MAX_TIMESTAMP specs should be failed here and the fulfillment stage
- // will later be retried for the potentially empty set of partitions with non-MAX_TIMESTAMP specs.
- if (maxTimestampPartitions.isEmpty()) {
+ // Otherwise, just the partitions with support specs should be failed here and the fulfillment stage
+ // will later be retried for the potentially empty set of partitions with non-support specs.
+ if (unsupportedVersionRetry == maxUnsupportedVersionRetry && accUnsupportedTimestampPartition.isEmpty()) {
Review Comment:
This condition suggests that when we've exhausted all retries _and_ accumulated no unsupported partitions, we fail everything. But if we've accumulated unsupported partitions during previous retries, this check doesn't trigger. Why not checking `supportedTimestampPartitions.isEmpty()`?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19706: ListOffsetsHandler should kick out the unsupported timestamp [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #20553:
URL: https://github.com/apache/kafka/pull/20553#discussion_r2412251186
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -202,23 +207,61 @@ private void handlePartitionError(
public Map<TopicPartition, Throwable> handleUnsupportedVersionException(
int brokerId, UnsupportedVersionException exception, Set<TopicPartition> keys
) {
- log.warn("Broker {} does not support MAX_TIMESTAMP offset specs", brokerId);
- Map<TopicPartition, Throwable> maxTimestampPartitions = new HashMap<>();
+ log.warn("Broker {} does not support {} offset specs", brokerId, timestampToString(currentUnsupportedVersion));
+ Map<TopicPartition, Throwable> supportedTimestampPartitions = new HashMap<>();
+
for (TopicPartition topicPartition : keys) {
Long offsetTimestamp = offsetTimestampsByPartition.get(topicPartition);
- if (offsetTimestamp == ListOffsetsRequest.MAX_TIMESTAMP) {
- maxTimestampPartitions.put(topicPartition, exception);
+ if (offsetTimestamp != null && offsetTimestamp == currentUnsupportedVersion && !accUnsupportedTimestampPartition.containsKey(topicPartition)) {
+ supportedTimestampPartitions.put(topicPartition, exception);
+ accUnsupportedTimestampPartition.put(topicPartition, exception);
}
}
- // If there are no partitions with MAX_TIMESTAMP specs the UnsupportedVersionException cannot be handled
+
+ // If there are no partitions with support specs the UnsupportedVersionException cannot be handled
// and all partitions should be failed here.
- // Otherwise, just the partitions with MAX_TIMESTAMP specs should be failed here and the fulfillment stage
- // will later be retried for the potentially empty set of partitions with non-MAX_TIMESTAMP specs.
- if (maxTimestampPartitions.isEmpty()) {
+ // Otherwise, just the partitions with support specs should be failed here and the fulfillment stage
+ // will later be retried for the potentially empty set of partitions with non-support specs.
+ if (unsupportedVersionRetry == maxUnsupportedVersionRetry && accUnsupportedTimestampPartition.isEmpty()) {
Review Comment:
We need to check the number of retry.
For example, there are 5 versions (-1, -2, -3, -4, -5), but we get the keys are [-1, -2, -3, -4], in such case, we still need to fall back next check instead of fail all.
I also add comments here, hope to make the intention is clear.
##########
clients/src/main/java/org/apache/kafka/clients/admin/internals/ListOffsetsHandler.java:
##########
@@ -202,23 +207,61 @@ private void handlePartitionError(
public Map<TopicPartition, Throwable> handleUnsupportedVersionException(
int brokerId, UnsupportedVersionException exception, Set<TopicPartition> keys
) {
- log.warn("Broker {} does not support MAX_TIMESTAMP offset specs", brokerId);
- Map<TopicPartition, Throwable> maxTimestampPartitions = new HashMap<>();
+ log.warn("Broker {} does not support {} offset specs", brokerId, timestampToString(currentUnsupportedVersion));
+ Map<TopicPartition, Throwable> supportedTimestampPartitions = new HashMap<>();
+
for (TopicPartition topicPartition : keys) {
Long offsetTimestamp = offsetTimestampsByPartition.get(topicPartition);
- if (offsetTimestamp == ListOffsetsRequest.MAX_TIMESTAMP) {
- maxTimestampPartitions.put(topicPartition, exception);
+ if (offsetTimestamp != null && offsetTimestamp == currentUnsupportedVersion && !accUnsupportedTimestampPartition.containsKey(topicPartition)) {
+ supportedTimestampPartitions.put(topicPartition, exception);
+ accUnsupportedTimestampPartition.put(topicPartition, exception);
Review Comment:
This is name mistake, fix it.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org