KAFKA-8983; AdminClient deleteRecords should not fail all partitions unnecessarily#7449
Merged
hachikuji merged 4 commits intoapache:trunkfrom Oct 8, 2019
Merged
KAFKA-8983; AdminClient deleteRecords should not fail all partitions unnecessarily#7449hachikuji merged 4 commits intoapache:trunkfrom
hachikuji merged 4 commits intoapache:trunkfrom
Conversation
3 tasks
ijuma
reviewed
Oct 4, 2019
| ExecutionException exception = assertThrows(ExecutionException.class, () -> future.get()); | ||
| assertTrue("Unexpected exception cause " + exception.getCause(), | ||
| exceptionClass.isAssignableFrom(exception.getCause().getClass())); | ||
| return (T) exception.getCause(); |
Member
There was a problem hiding this comment.
I think you can avoid the warning (and suppression) if you use exceptionClass.cast.
Contributor
Author
There was a problem hiding this comment.
Woah. Didn't know about that.
ijuma
reviewed
Oct 4, 2019
| public static <T extends Throwable> T assertFutureThrows(KafkaFuture<?> future, Class<T> exceptionClass) { | ||
| ExecutionException exception = assertThrows(ExecutionException.class, () -> future.get()); | ||
| assertTrue("Unexpected exception cause " + exception.getCause(), | ||
| exceptionClass.isAssignableFrom(exception.getCause().getClass())); |
ijuma
approved these changes
Oct 6, 2019
| return tps; | ||
| } | ||
|
|
||
| public static <T extends Throwable> T assertFutureThrows(KafkaFuture<?> future, Class<T> exceptionClass) { |
Member
There was a problem hiding this comment.
Nit: shall we call the parameter exceptionCauseClass and add brief javadoc saying that this checks the cause of the execution exception?
| @Override | ||
| public void close() { | ||
| this.adminClient.close(); | ||
| this.adminClient.close(Duration.ZERO); |
Member
There was a problem hiding this comment.
The default implementation of close() is to wait forever. What's the reasoning for using Duration.ZERO here?
Contributor
Author
There was a problem hiding this comment.
Indefinite blocking seems like a bad combination when time is often mocked.
Contributor
Author
There was a problem hiding this comment.
That said, I don't think I actually need this change for this patch, so I can revert it.
| final long nowDelete = time.milliseconds(); | ||
|
|
||
| final Map<TopicPartition, Long> partitionDeleteOffsets = entry.getValue(); | ||
| final long currentTimeMs = time.milliseconds(); |
Member
There was a problem hiding this comment.
An existing issue, but would it be better to do this outside the loop?
| } | ||
| } | ||
|
|
||
| for (final Map.Entry<Node, Map<TopicPartition, Long>> entry: leaders.entrySet()) { |
Member
There was a problem hiding this comment.
Nit: existing issue, but can we add a space before :?
hachikuji
added a commit
that referenced
this pull request
Oct 8, 2019
…unnecessarily (#7449) The deleteRecords API in the AdminClient groups records to be sent by the partition leaders. If one of these requests fails, we currently fail all futures, including those tied to requests sent to other leaders. It would be better to fail only those partitions included in the failed request. Reviewers: Ismael Juma <[email protected]>
ijuma
added a commit
to ijuma/kafka
that referenced
this pull request
Apr 28, 2020
…t-for-generated-requests * apache-github/trunk: KAFKA-8932; Add tag for CreateTopicsResponse.TopicConfigErrorCode (KIP-525) (apache#7464) KAFKA-8944: Fixed KTable compiler warning. (apache#7393) KAFKA-8964: Rename tag client-id for thread-level metrics and below (apache#7429) MINOR: remove unused imports in Streams system tests (apache#7468) KAFKA-7190; Retain producer state until transactionalIdExpiration time passes (apache#7388) KAFKA-8983; AdminClient deleteRecords should not fail all partitions unnecessarily (apache#7449) MINOR: Modified Exception handling for KIP-470 (apache#7461) KAFKA-7245: Deprecate WindowStore#put(key, value) (apache#7105) KAFKA-8179: Part 7, cooperative rebalancing in Streams (apache#7386) KAFKA-8985; Add flexible version support to inter-broker APIs (apache#7453) MINOR: Bump version to 2.5.0-SNAPSHOT (apache#7455)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
The deleteRecords API in the AdminClient groups records to be sent by the partition leaders. If one of these requests fails, we currently fail all futures, including those tied to requests sent to other leaders. It would be better to fail only those partitions included in the failed request.
Committer Checklist (excluded from commit message)