Skip to content
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

Jump to bottom

MINOR: Eliminate warnings for AdminUtils #16470

Merged
merged 12 commits into from
Jun 30, 2024
Merged

MINOR: Eliminate warnings for AdminUtils #16470

Changes from 4 commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
aec8cde
MINOR: Eliminate warnings for AdminUtils
LoganZhuZzz Jun 27, 2024
73bca12
Merge branch 'apache:trunk' into trunk
LoganZhuZzz Jun 27, 2024
b2e3714
MINOR: Fix build issues
LoganZhuZzz Jun 27, 2024
c50d997
Merge branch 'apache:trunk' into trunk
LoganZhuZzz Jun 27, 2024
6a3b3d5
MINOR: Original code is straightforward enough, revert and refactor.
LoganZhuZzz Jun 27, 2024
2d06142
Merge branch 'apache:trunk' into trunk
LoganZhuZzz Jun 28, 2024
1ad9d29
Merge branch 'apache:trunk' into trunk
LoganZhuZzz Jun 28, 2024
87e1734
MINOR: Revert unrelated changes
LoganZhuZzz Jun 28, 2024
1e35470
Merge branch 'apache:trunk' into trunk
LoganZhuZzz Jun 28, 2024
6e36429
Merge branch 'apache:trunk' into trunk
LoganZhuZzz Jun 29, 2024
eaf9a5e
Merge branch 'apache:trunk' into trunk
LoganZhuZzz Jun 30, 2024
e2ff23c
MINOR: Add unit test for AdminUtils
LoganZhuZzz Jun 30, 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
24 changes: 19 additions & 5 deletions server-common/src/main/java/org/apache/kafka/admin/AdminUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -143,9 +143,9 @@ private static Map<Integer, List<Integer>> assignReplicasToBrokersRackUnaware(in
int fixedStartIndex,
int startPartitionId) {
Map<Integer, List<Integer>> ret = new HashMap<>();
int startIndex = fixedStartIndex >= 0 ? fixedStartIndex : RAND.nextInt(brokerList.size());
int startIndex = determineIndex(fixedStartIndex, brokerList.size());
int currentPartitionId = Math.max(0, startPartitionId);
int nextReplicaShift = fixedStartIndex >= 0 ? fixedStartIndex : RAND.nextInt(brokerList.size());
int nextReplicaShift = determineIndex(fixedStartIndex, brokerList.size());
Copy link
Contributor

Choose a reason for hiding this comment

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

I don't think we need to refactor this function,the original code is straightforward enough

for (int i = 0; i < nPartitions; i++) {
if (currentPartitionId > 0 && (currentPartitionId % brokerList.size() == 0))
nextReplicaShift += 1;
Expand All @@ -166,14 +166,14 @@ private static Map<Integer, List<Integer>> assignReplicasToBrokersRackAware(int
int fixedStartIndex,
int startPartitionId) {
Map<Integer, String> brokerRackMap = new HashMap<>();
brokerMetadatas.forEach(m -> brokerRackMap.put(m.id, m.rack.get()));
brokerMetadatas.forEach(m -> brokerRackMap.put(m.id, m.rack.orElse(null)));
Copy link
Contributor

Choose a reason for hiding this comment

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

Is it expected that rack will be empty

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Is it expected that rack will be empty

Before calling this function, the parameter brokerMetadatas has already been checked. If rack is empty, an AdminOperationException will be thrown. However, my IDE still issues a warning, so I made an equivalent replacement. Next time, I will ignore this situation.

Copy link
Contributor

Choose a reason for hiding this comment

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

Before calling this function, the parameter brokerMetadatas has already been checked. If rack is empty, an AdminOperationException will be thrown. However, my IDE still issues a warning, so I made an equivalent replacement. Next time, I will ignore this situation.

How about moving the check into assignReplicasToBrokersRackAware? for example:

brokerMetadatas.forEach(m -> brokerRackMap.put(m.id, m.rack.orElseThrow(() -> new AdminOperationException("Not all brokers have rack information for replica rack aware assignment."))));

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Before calling this function, the parameter brokerMetadatas has already been checked. If rack is empty, an AdminOperationException will be thrown. However, my IDE still issues a warning, so I made an equivalent replacement. Next time, I will ignore this situation.

How about moving the check into assignReplicasToBrokersRackAware? for example:

brokerMetadatas.forEach(m -> brokerRackMap.put(m.id, m.rack.orElseThrow(() -> new AdminOperationException("Not all brokers have rack information for replica rack aware assignment."))));

great suggestion, I will do it

int numRacks = new HashSet<>(brokerRackMap.values()).size();
List<Integer> arrangedBrokerList = getRackAlternatedBrokerList(brokerRackMap);
int numBrokers = arrangedBrokerList.size();
Map<Integer, List<Integer>> ret = new HashMap<>();
int startIndex = fixedStartIndex >= 0 ? fixedStartIndex : RAND.nextInt(arrangedBrokerList.size());
int startIndex = determineIndex(fixedStartIndex, arrangedBrokerList.size());
int currentPartitionId = Math.max(0, startPartitionId);
int nextReplicaShift = fixedStartIndex >= 0 ? fixedStartIndex : RAND.nextInt(arrangedBrokerList.size());
int nextReplicaShift = determineIndex(fixedStartIndex, arrangedBrokerList.size());
for (int i = 0; i < nPartitions; i++) {
if (currentPartitionId > 0 && (currentPartitionId % arrangedBrokerList.size() == 0))
nextReplicaShift += 1;
Expand Down Expand Up @@ -211,6 +211,20 @@ private static Map<Integer, List<Integer>> assignReplicasToBrokersRackAware(int
return ret;
}

/**
* Determines the index value.
*
* If the provided fixed index value is greater than or equal to 0, returns the fixed index.
* Otherwise, generates a random index between 0 (inclusive) and the broker size (exclusive).
*
* @param fixedIndex the fixed index value, used if it is greater than or equal to 0
* @param brokerSize the size of the broker list, used as the upper limit for generating a random index
* @return the calculated index value
*/
private static int determineIndex(int fixedIndex, int brokerSize) {
return fixedIndex >= 0 ? fixedIndex : RAND.nextInt(brokerSize);
}

/**
* Given broker and rack information, returns a list of brokers alternated by the rack. Assume
* this is the rack and its brokers:
Expand Down