jolshan commented on code in PR #13247:
URL: https://github.com/apache/kafka/pull/13247#discussion_r1350759368
##########
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsIntegrationTest.java:
##########
@@ -857,68 +864,18 @@ public void close() {
}
}
- private ReassignPartitionsCommand.VerifyAssignmentResult
asScala(VerifyAssignmentResult res) {
- Map<TopicPartition,
ReassignPartitionsCommand.PartitionReassignmentState> partStates = new
HashMap<>();
- res.partStates.forEach((tp, state) -> partStates.put(tp,
asScala(state)));
-
- Map<TopicPartitionReplica, ReassignPartitionsCommand.LogDirMoveState>
moveStates = new HashMap<>();
- res.moveStates.forEach((tpr, state) -> moveStates.put(tpr,
asScala(state)));
-
- return new
ReassignPartitionsCommand.VerifyAssignmentResult(asScala(partStates),
res.partsOngoing, asScala(moveStates), res.movesOngoing);
- }
-
- @SuppressWarnings({"unchecked"})
- private ReassignPartitionsCommand.PartitionReassignmentState
asScala(PartitionReassignmentState state) {
- return new ReassignPartitionsCommand.PartitionReassignmentState(
- seq((List) state.currentReplicas),
- seq((List) state.targetReplicas),
- state.done
- );
- }
-
- private ReassignPartitionsCommand.LogDirMoveState asScala(LogDirMoveState
state) {
- if (state instanceof ActiveMoveState) {
- ActiveMoveState s = (ActiveMoveState) state;
- return new
ReassignPartitionsCommand.ActiveMoveState(s.currentLogDir, s.targetLogDir,
s.futureLogDir);
- } else if (state instanceof CancelledMoveState) {
- CancelledMoveState s = (CancelledMoveState) state;
- return new
ReassignPartitionsCommand.CancelledMoveState(s.currentLogDir, s.targetLogDir);
- } else if (state instanceof CompletedMoveState) {
- CompletedMoveState s = (CompletedMoveState) state;
- return new
ReassignPartitionsCommand.CompletedMoveState(s.targetLogDir);
- } else if (state instanceof MissingLogDirMoveState) {
- MissingLogDirMoveState s = (MissingLogDirMoveState) state;
- return new
ReassignPartitionsCommand.MissingLogDirMoveState(s.targetLogDir);
- } else if (state instanceof MissingReplicaMoveState) {
- MissingReplicaMoveState s = (MissingReplicaMoveState) state;
- return new
ReassignPartitionsCommand.MissingReplicaMoveState(s.targetLogDir);
- }
-
- throw new IllegalArgumentException("Unknown state " + state);
- }
-
@SuppressWarnings("unchecked")
- static <T> scala.collection.immutable.Set<T> set(final T... set) {
- return mutableSet(set).toSet();
+ static <T> Set<T> set(final T... set) {
+ return new HashSet<>(Arrays.asList(set));
Review Comment:
Got it. Thanks for clarifying. (That's going to be a fun one to migrate too
😅 )
##########
tools/src/test/java/org/apache/kafka/tools/reassign/ReassignPartitionsIntegrationTest.java:
##########
@@ -857,68 +864,18 @@ public void close() {
}
}
- private ReassignPartitionsCommand.VerifyAssignmentResult
asScala(VerifyAssignmentResult res) {
- Map<TopicPartition,
ReassignPartitionsCommand.PartitionReassignmentState> partStates = new
HashMap<>();
- res.partStates.forEach((tp, state) -> partStates.put(tp,
asScala(state)));
-
- Map<TopicPartitionReplica, ReassignPartitionsCommand.LogDirMoveState>
moveStates = new HashMap<>();
- res.moveStates.forEach((tpr, state) -> moveStates.put(tpr,
asScala(state)));
-
- return new
ReassignPartitionsCommand.VerifyAssignmentResult(asScala(partStates),
res.partsOngoing, asScala(moveStates), res.movesOngoing);
- }
-
- @SuppressWarnings({"unchecked"})
- private ReassignPartitionsCommand.PartitionReassignmentState
asScala(PartitionReassignmentState state) {
- return new ReassignPartitionsCommand.PartitionReassignmentState(
- seq((List) state.currentReplicas),
- seq((List) state.targetReplicas),
- state.done
- );
- }
-
- private ReassignPartitionsCommand.LogDirMoveState asScala(LogDirMoveState
state) {
- if (state instanceof ActiveMoveState) {
- ActiveMoveState s = (ActiveMoveState) state;
- return new
ReassignPartitionsCommand.ActiveMoveState(s.currentLogDir, s.targetLogDir,
s.futureLogDir);
- } else if (state instanceof CancelledMoveState) {
- CancelledMoveState s = (CancelledMoveState) state;
- return new
ReassignPartitionsCommand.CancelledMoveState(s.currentLogDir, s.targetLogDir);
- } else if (state instanceof CompletedMoveState) {
- CompletedMoveState s = (CompletedMoveState) state;
- return new
ReassignPartitionsCommand.CompletedMoveState(s.targetLogDir);
- } else if (state instanceof MissingLogDirMoveState) {
- MissingLogDirMoveState s = (MissingLogDirMoveState) state;
- return new
ReassignPartitionsCommand.MissingLogDirMoveState(s.targetLogDir);
- } else if (state instanceof MissingReplicaMoveState) {
- MissingReplicaMoveState s = (MissingReplicaMoveState) state;
- return new
ReassignPartitionsCommand.MissingReplicaMoveState(s.targetLogDir);
- }
-
- throw new IllegalArgumentException("Unknown state " + state);
- }
-
@SuppressWarnings("unchecked")
- static <T> scala.collection.immutable.Set<T> set(final T... set) {
- return mutableSet(set).toSet();
+ static <T> Set<T> set(final T... set) {
+ return new HashSet<>(Arrays.asList(set));
Review Comment:
Got it. Thanks for clarifying. (That's going to be a fun one to migrate too
😅 )
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
