adoroszlai commented on code in PR #11805:
URL: https://github.com/apache/ozone/pull/11805#discussion_r4445166823
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ECUnderReplicationHandler.java:
##########
@@ -144,15 +143,11 @@ public int processAndSendCommands(
}
}
- Map<Integer, Pair<ContainerReplica, NodeStatus>> sources =
- filterSources(replicas, deletionInFlight);
- List<DatanodeDetails> availableSourceNodes =
- sources.values().stream().map(Pair::getLeft)
- .map(ContainerReplica::getDatanodeDetails)
- .filter(datanodeDetails ->
- datanodeDetails.getPersistedOpState() ==
- HddsProtos.NodeOperationalState.IN_SERVICE)
- .collect(Collectors.toList());
+ Map<Integer, SourceReplica> sources = filterSources(replicas,
deletionInFlight);
+ List<DatanodeDetails> availableSourceNodes =
sources.values().stream().map(SourceReplica::replica)
+ .map(ContainerReplica::getDatanodeDetails).filter(datanodeDetails ->
+ datanodeDetails.getPersistedOpState() ==
HddsProtos.NodeOperationalState.IN_SERVICE)
+ .collect(Collectors.toList());
Review Comment:
nit: please format streams for easier reading (one operation per line)
```suggestion
List<DatanodeDetails> availableSourceNodes = sources.values().stream()
.map(SourceReplica::replica)
.map(ContainerReplica::getDatanodeDetails)
.filter(dn -> dn.getPersistedOpState() ==
HddsProtos.NodeOperationalState.IN_SERVICE)
.collect(Collectors.toList());
```
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ECUnderReplicationHandler.java:
##########
@@ -450,12 +440,9 @@ private int processReconstruction(
return commandsSent;
}
- private static boolean hasSufficientSourcesForDecommissionReconstruction(
- Map<Integer, Pair<ContainerReplica, NodeStatus>> sources,
- Set<Integer> decomIndexes,
- int requiredDataSources) {
- long nonDecommissionSourceCount = sources.keySet().stream()
- .filter(index -> !decomIndexes.contains(index))
+ private static boolean
hasSufficientSourcesForDecommissionReconstruction(Map<Integer, SourceReplica>
sources,
+ Set<Integer> decomIndexes, int requiredDataSources) {
+ long nonDecommissionSourceCount = sources.keySet().stream().filter(index
-> !decomIndexes.contains(index))
Review Comment:
```suggestion
long nonDecommissionSourceCount = sources.keySet().stream()
.filter(index -> !decomIndexes.contains(index))
```
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ECUnderReplicationHandler.java:
##########
@@ -513,9 +498,8 @@ private int processDecommissioningIndexes(
metrics.incrEcReconstructionGlobalLimitReachedTotal();
break;
}
- LOG.info("Source node {} is highly loaded, switching to "
- + "reconstruction for decommissioning container {}",
- source.getLeft().getDatanodeDetails(), container.containerID());
+ LOG.info("Source node {} is highly loaded, switching to " +
"reconstruction for decommissioning container {}",
Review Comment:
```suggestion
LOG.info("Source node {} is highly loaded, switching to
reconstruction for decommissioning container {}",
```
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ECUnderReplicationHandler.java:
##########
@@ -239,25 +234,22 @@ private Map<Integer, Pair<ContainerReplica, NodeStatus>>
filterSources(
.filter(r -> !deletionInFlight.contains(r.getDatanodeDetails()))
.map(r -> {
try {
- return Pair.of(r,
- replicationManager.getNodeStatus(r.getDatanodeDetails()));
+ return new SourceReplica(r,
replicationManager.getNodeStatus(r.getDatanodeDetails()));
} catch (NodeNotFoundException e) {
throw new IllegalStateException("Unable to find NodeStatus for "
+ r.getDatanodeDetails(), e);
}
})
- .filter(pair -> pair.getRight().isHealthy())
+ .filter(source -> source.nodeStatus().isHealthy())
// If there are multiple nodes online for a given index, we just
// pick any IN_SERVICE one. At the moment, the input streams cannot
// handle multiple replicas for the same index, so if we passed them
// all through they would not get used anyway.
// If neither of the nodes are in service, we just pass one through,
// as it will be decommission or maintenance.
- .collect(Collectors.toMap(
- pair -> pair.getLeft().getReplicaIndex(),
- pair -> pair,
- (p1, p2) -> {
- if (p1.getRight().getOperationalState() == IN_SERVICE) {
+ .collect(Collectors.toMap(source -> source.replica().getReplicaIndex(),
+ source -> source, (p1, p2) -> {
Review Comment:
```suggestion
.collect(Collectors.toMap(
source -> source.replica().getReplicaIndex(),
source -> source,
(p1, p2) -> {
```
##########
hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationTestUtil.java:
##########
@@ -72,19 +71,18 @@ private ReplicationTestUtil() {
}
@SafeVarargs
Review Comment:
With generic `Pair<...>` gone, I don't think `@SafeVarargs` is needed
anymore.
```suggestion
```
Also in other methods where varargs parameter is no longer generic.
--
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]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
