Skip to content

Commit

Permalink
KAFKA-16828: RackAwareTaskAssignorTest failed
Browse files Browse the repository at this point in the history
  • Loading branch information
@brandboat
brandboat committed May 23, 2024
1 parent 14b5c4d commit fb3ad0b
Show file tree
Hide file tree
Showing 2 changed files with 18 additions and 16 deletions.
4 changes: 2 additions & 2 deletions ...n/java/org/apache/kafka/streams/processor/internals/assignment/RackAwareTaskAssignor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -139,7 +139,7 @@ public boolean populateTopicsToDescribe(final Set<String> topicsToDescribe,
return populateTopicsToDescribe(fullMetadata, changelogPartitionsForTask, partitionsForTask, changelog, topicsToDescribe, racksForPartition);
}

public static boolean populateTopicsToDescribe(final Cluster fullMetadata,
public boolean populateTopicsToDescribe(final Cluster fullMetadata,
final Map<TaskId, Set<TopicPartition>> changelogPartitionsForTask,
final Map<TaskId, Set<TopicPartition>> partitionsForTask,
final boolean changelog,
Expand Down Expand Up @@ -188,7 +188,7 @@ private boolean validateTopicPartitionRack(final boolean changelogTopics) {
*
* @return whether the operation successfully completed and the rack information is valid.
*/
public static boolean validateTopicPartitionRack(final Cluster fullMetadata,
public boolean validateTopicPartitionRack(final Cluster fullMetadata,
final InternalTopicManager internalTopicManager,
final Map<TaskId, Set<TopicPartition>> changelogPartitionsForTask,
final Map<TaskId, Set<TopicPartition>> partitionsForTask,
Expand Down
30 changes: 16 additions & 14 deletions ...va/org/apache/kafka/streams/processor/internals/assignment/RackAwareTaskAssignorTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -76,6 +76,8 @@
import static org.junit.Assert.assertTrue;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyMap;
import static org.mockito.ArgumentMatchers.anySet;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doReturn;
Expand Down Expand Up @@ -194,8 +196,8 @@ public void shouldDisableAssignorFromConfig() {

// False since partitionWithoutInfo10 is missing in cluster metadata
assertFalse(assignor.canEnableRackAwareAssignor());
verify(assignor, never()).populateTopicsToDescribe(anySet(), eq(false));
verify(assignor, never()).populateTopicsToDescribe(anySet(), eq(true));
verify(assignor, never()).populateTopicsToDescribe(any(), anyMap(), anyMap(), eq(false), anySet(), anyMap());
verify(assignor, never()).populateTopicsToDescribe(any(), anyMap(), anyMap(), eq(true), anySet(), anyMap());
}

@Test
Expand All @@ -213,8 +215,8 @@ public void shouldDisableActiveWhenMissingClusterInfo() {

// False since partitionWithoutInfo10 is missing in cluster metadata
assertFalse(assignor.canEnableRackAwareAssignor());
verify(assignor, times(1)).populateTopicsToDescribe(anySet(), eq(false));
verify(assignor, never()).populateTopicsToDescribe(anySet(), eq(true));
verify(assignor, times(1)).populateTopicsToDescribe(any(), anyMap(), anyMap(), eq(false), anySet(), anyMap());
verify(assignor, never()).populateTopicsToDescribe(any(), anyMap(), anyMap(), eq(true), anySet(), anyMap());
assertFalse(assignor.populateTopicsToDescribe(new HashSet<>(), false));
}

Expand All @@ -233,8 +235,8 @@ public void shouldDisableActiveWhenRackMissingInNode() {

// False since nodeMissingRack has one node which doesn't have rack
assertFalse(assignor.canEnableRackAwareAssignor());
verify(assignor, times(1)).populateTopicsToDescribe(anySet(), eq(false));
verify(assignor, never()).populateTopicsToDescribe(anySet(), eq(true));
verify(assignor, times(1)).populateTopicsToDescribe(any(), anyMap(), anyMap(), eq(false), anySet(), anyMap());
verify(assignor, never()).populateTopicsToDescribe(any(), anyMap(), anyMap(), eq(true), anySet(), anyMap());
assertFalse(assignor.populateTopicsToDescribe(new HashSet<>(), false));
}

Expand Down Expand Up @@ -342,12 +344,12 @@ public void shouldEnableRackAwareAssignorWithCacheResult() {

// partitionWithoutInfo00 has rackInfo in cluster metadata
assertTrue(assignor.canEnableRackAwareAssignor());
verify(assignor, times(1)).populateTopicsToDescribe(anySet(), eq(false));
verify(assignor, times(1)).populateTopicsToDescribe(any(), anyMap(), anyMap(), eq(false), anySet(), anyMap());

// Should use cache result
Mockito.reset(assignor);
assertTrue(assignor.canEnableRackAwareAssignor());
verify(assignor, never()).populateTopicsToDescribe(anySet(), eq(false));
verify(assignor, never()).populateTopicsToDescribe(any(), anyMap(), anyMap(), eq(false), anySet(), anyMap());
}

@Test
Expand Down Expand Up @@ -409,8 +411,8 @@ public void shouldEnableRackAwareAssignorWithStandbyDescribingTopics() {
));

assertTrue(assignor.canEnableRackAwareAssignor());
verify(assignor, times(1)).populateTopicsToDescribe(anySet(), eq(false));
verify(assignor, times(1)).populateTopicsToDescribe(anySet(), eq(true));
verify(assignor, times(1)).populateTopicsToDescribe(any(), anyMap(), anyMap(), eq(false), anySet(), anyMap());
verify(assignor, times(1)).populateTopicsToDescribe(any(), anyMap(), anyMap(), eq(true), anySet(), anyMap());

final Map<TopicPartition, Set<String>> racksForPartition = assignor.racksForPartition();
final Map<TopicPartition, Set<String>> expected = mkMap(
Expand Down Expand Up @@ -447,8 +449,8 @@ public void shouldDisableRackAwareAssignorWithStandbyDescribingTopicsFailure() {
));

assertFalse(assignor.canEnableRackAwareAssignor());
verify(assignor, times(1)).populateTopicsToDescribe(anySet(), eq(false));
verify(assignor, times(1)).populateTopicsToDescribe(anySet(), eq(true));
verify(assignor, times(1)).populateTopicsToDescribe(any(), anyMap(), anyMap(), eq(false), anySet(), anyMap());
verify(assignor, times(1)).populateTopicsToDescribe(any(), anyMap(), anyMap(), eq(true), anySet(), anyMap());
}

@Test
Expand All @@ -469,8 +471,8 @@ public void shouldDisableRackAwareAssignorWithDescribingTopicsFailure() {
));

assertFalse(assignor.canEnableRackAwareAssignor());
verify(assignor, times(1)).populateTopicsToDescribe(anySet(), eq(false));
verify(assignor, never()).populateTopicsToDescribe(anySet(), eq(true));
verify(assignor, times(1)).populateTopicsToDescribe(any(), anyMap(), anyMap(), eq(false), anySet(), anyMap());
verify(assignor, never()).populateTopicsToDescribe(any(), anyMap(), anyMap(), eq(true), anySet(), anyMap());
assertTrue(assignor.populateTopicsToDescribe(new HashSet<>(), false));
}

Expand Down

0 comments on commit fb3ad0b

Please sign in to comment.