yifan-c commented on code in PR #30:
URL:
https://github.com/apache/cassandra-analytics/pull/30#discussion_r1447772924
##########
cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/RecordWriterTest.java:
##########
@@ -120,6 +122,34 @@ public void testWriteFailWhenTopologyChangeWithinTask()
assertThat(ex.getMessage(), endsWith("Token range mappings have
changed since the task started"));
}
+ @Test
+ public void testWriteWithBlockedInstances()
Review Comment:
remove extra spaces
##########
cassandra-analytics-core/src/test/java/org/apache/cassandra/spark/bulkwriter/TokenRangeMappingUtils.java:
##########
@@ -54,6 +54,41 @@ public static TokenRangeMapping<RingInstance>
buildTokenRangeMapping(final int i
return buildTokenRangeMapping(initialToken, rfByDC, instancesPerDC,
false, -1);
}
+ public static TokenRangeMapping<RingInstance>
buildTokenRangeMappingWithBlockedInstance(int initialToken,
+
ImmutableMap<String, Integer> rfByDC,
+
int instancesPerDC)
+ {
+ final List<RingInstance> instances = getInstances(initialToken,
rfByDC, instancesPerDC);
Review Comment:
remove `final`
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/BulkWriteValidator.java:
##########
@@ -122,27 +122,43 @@ private void updateInstanceAvailability()
private void validateAvailabilityAndUpdateFailures(RingInstance instance,
InstanceAvailability availability)
{
- if (availability == InstanceAvailability.INVALID_STATE)
+ switch (availability)
{
- // If we find any nodes in a totally invalid state, just throw as
we can't continue
- String message = String.format("Instance (%s) is in an invalid
state (%s) during import. "
- + "Please rerun import once
topology changes are complete.",
- instance.getNodeName(),
cluster.getInstanceState(instance));
- throw new RuntimeException(message);
- }
-
- if (availability == InstanceAvailability.UNAVAILABLE_BLOCKED
- || availability == InstanceAvailability.UNAVAILABLE_DOWN)
- {
- Collection<Range<BigInteger>> failedRanges =
cluster.getTokenRangeMapping(true)
-
.getTokenRanges()
- .get(instance);
- failedRanges.forEach(failedRange -> {
- String nodeDisplayName = instance.getNodeName();
- String message = String.format("%s %s", nodeDisplayName,
availability.getMessage());
- LOGGER.warn("{} failed in phase {} on {} because {}",
failedRange, phase, nodeDisplayName, message);
- failureHandler.addFailure(failedRange, instance, message);
- });
+ case INVALID_STATE:
+ // If we find any nodes in a totally invalid state, just throw
as we can't continue
+ String errorMessage = String.format("Instance (%s) is in an
invalid state (%s) during import. "
+ + "Please rerun import
once topology changes are complete.",
+ instance.getNodeName(),
cluster.getInstanceState(instance));
+ throw new RuntimeException(errorMessage);
+
+ // Check for blocked instances and ranges for the purpose of
logging only.
+ // We check for blocked instances while validating consistency
level requirements
+ case UNAVAILABLE_BLOCKED:
+ Collection<Range<BigInteger>> rangesInBlockedInstance =
cluster.getTokenRangeMapping(true)
+
.getTokenRanges()
+
.get(instance);
+ rangesInBlockedInstance.forEach(failedRange -> {
+ String nodeDisplayName = instance.getNodeName();
+ String message = String.format("%s %s", nodeDisplayName,
availability.getMessage());
+ LOGGER.warn("{} failed in phase {} on {} because {}",
failedRange, phase, nodeDisplayName, message);
+ });
+ break;
+
+ case UNAVAILABLE_DOWN:
+ Collection<Range<BigInteger>> failedRanges =
cluster.getTokenRangeMapping(true)
+
.getTokenRanges()
+
.get(instance);
+ failedRanges.forEach(failedRange -> {
+ String nodeDisplayName = instance.getNodeName();
+ String message = String.format("%s %s", nodeDisplayName,
availability.getMessage());
+ LOGGER.warn("{} failed in phase {} on {} because {}",
failedRange, phase, nodeDisplayName, message);
+ failureHandler.addFailure(failedRange, instance, message);
+ });
+ break;
Review Comment:
nit: to remove some duplication
```suggestion
case UNAVAILABLE_BLOCKED:
case UNAVAILABLE_DOWN:
boolean shouldAddFailure = availability ==
InstanceAvailability.UNAVAILABLE_DOWN;
Collection<Range<BigInteger>> unavailableRanges =
cluster.getTokenRangeMapping(true)
.getTokenRanges()
.get(instance);
unavailableRanges.forEach(failedRange -> {
String nodeDisplayName = instance.getNodeName();
String message = String.format("%s %s", nodeDisplayName,
availability.getMessage());
LOGGER.warn("{} failed in phase {} on {} because {}",
failedRange, phase, nodeDisplayName, message);
if (shouldAddFailure)
{
failureHandler.addFailure(failedRange, instance,
message);
}
});
break;
```
--
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: commits-unsubscr...@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@cassandra.apache.org
For additional commands, e-mail: commits-h...@cassandra.apache.org
