Skip to content

KAFKA-16828: RackAwareTaskAssignorTest failed #16044

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
chia7712 merged 4 commits into from
May 23, 2024
Merged

KAFKA-16828: RackAwareTaskAssignorTest failed #16044

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
fb3ad0b
KAFKA-16828: RackAwareTaskAssignorTest failed
brandboat May 23, 2024
73c2aa1
Revert "KAFKA-16828: RackAwareTaskAssignorTest failed"
brandboat May 23, 2024
e04ad1d
Address comments
brandboat May 23, 2024
61ccb83
Use private classifier
brandboat May 23, 2024
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
25 changes: 3 additions & 22 deletions ...n/java/org/apache/kafka/streams/processor/internals/assignment/RackAwareTaskAssignor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,17 +134,7 @@ public synchronized boolean canEnableRackAwareAssignor() {
}

// Visible for testing. This method also checks if all TopicPartitions exist in cluster
public boolean populateTopicsToDescribe(final Set<String> topicsToDescribe,
final boolean changelog) {
return populateTopicsToDescribe(fullMetadata, changelogPartitionsForTask, partitionsForTask, changelog, topicsToDescribe, racksForPartition);
}

public static boolean populateTopicsToDescribe(final Cluster fullMetadata,

@chia7712 chia7712 May 23, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ditto.

@chia7712 chia7712 May 23, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I feel addressing this comment can fix all failed tests.

@brandboat brandboat May 23, 2024

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks, the solution is cleaner.

final Map<TaskId, Set<TopicPartition>> changelogPartitionsForTask,
final Map<TaskId, Set<TopicPartition>> partitionsForTask,
final boolean changelog,
final Set<String> topicsToDescribe,
final Map<TopicPartition, Set<String>> racksForPartition) {
boolean populateTopicsToDescribe(final Set<String> topicsToDescribe, final boolean changelog) {
if (changelog) {
// Changelog topics are not in metadata, we need to describe them
changelogPartitionsForTask.values().stream().flatMap(Collection::stream).forEach(tp -> topicsToDescribe.add(tp.topic()));
Expand Down Expand Up @@ -177,26 +167,17 @@ public static boolean populateTopicsToDescribe(final Cluster fullMetadata,
return true;
}

private boolean validateTopicPartitionRack(final boolean changelogTopics) {
return validateTopicPartitionRack(fullMetadata, internalTopicManager, changelogPartitionsForTask, partitionsForTask, changelogTopics, racksForPartition);
}

/**
* This function populates the {@param racksForPartition} parameter passed into the function by using both
* the {@code Cluster} metadata as well as the {@param internalTopicManager} for topics that have stale
* information.
*
* @return whether the operation successfully completed and the rack information is valid.
*/
public static boolean validateTopicPartitionRack(final Cluster fullMetadata,
final InternalTopicManager internalTopicManager,
final Map<TaskId, Set<TopicPartition>> changelogPartitionsForTask,
final Map<TaskId, Set<TopicPartition>> partitionsForTask,
final boolean changelogTopics,
final Map<TopicPartition, Set<String>> racksForPartition) {
private boolean validateTopicPartitionRack(final boolean changelogTopics) {
// Make sure rackId exist for all TopicPartitions needed
final Set<String> topicsToDescribe = new HashSet<>();
if (!populateTopicsToDescribe(fullMetadata, changelogPartitionsForTask, partitionsForTask, changelogTopics, topicsToDescribe, racksForPartition)) {
if (!populateTopicsToDescribe(topicsToDescribe, changelogTopics)) {
return false;
}

Expand Down