lukasz-antoniak commented on code in PR #239:
URL:
https://github.com/apache/cassandra-analytics/pull/239#discussion_r3911834501
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSession.java:
##########
@@ -182,57 +171,59 @@ protected void sendRemainingSSTables()
}
}
- private void sendSStableToReplicas(Path dataFile)
+ private void sendSStableToReplicas(SortedSSTableWriter.PreparedSSTable
preparedSSTable)
{
int ssTableIdx = nextSSTableIdx.getAndIncrement();
LOGGER.info("[{}]: Pushing SSTable {} to replicas {}",
- sessionID, dataFile,
+ sessionID, preparedSSTable.dataFile(),
replicas.stream().map(RingInstance::nodeName).collect(Collectors.joining(",")));
- replicas.removeIf(replica -> !trySendSSTableToOneReplica(dataFile,
ssTableIdx, replica, sstableWriter.fileDigestMap()));
+ replicas.removeIf(replica ->
!trySendSSTableToOneReplica(preparedSSTable, ssTableIdx, replica));
}
- private boolean trySendSSTableToOneReplica(Path dataFile,
+ private boolean
trySendSSTableToOneReplica(SortedSSTableWriter.PreparedSSTable preparedSSTable,
int ssTableIdx,
- RingInstance replica,
- Map<Path, Digest> fileDigests)
+ RingInstance replica)
{
try
{
- sendSSTableToOneReplica(dataFile, ssTableIdx, replica,
fileDigests);
+ sendSSTableToOneReplica(preparedSSTable, ssTableIdx, replica);
return true;
}
catch (Exception exception)
{
LOGGER.error("[{}]: Failed to stream range {} to instance {}",
sessionID, tokenRange, replica.nodeName(), exception);
writerContext.cluster().refreshClusterInfo();
- failureHandler.addFailure(this.tokenRange, replica,
exception.getMessage());
- errors.add(new StreamError(this.tokenRange, replica,
exception.getMessage()));
+ // Sometimes error can contain just file name (e.g. when it is
missing).
+ // Let us return 3 latest stacktrace lines for easier
troubleshooting.
+ String stackTrace = Arrays.stream(exception.getStackTrace())
+ .limit(3)
+ .map(StackTraceElement::toString)
+ .collect(Collectors.joining("\n"));
+ String errorMessage = exception.getClass().getName() + ": " +
exception.getMessage()
+ + "\n" + String.join("\n", stackTrace);
+ failureHandler.addFailure(this.tokenRange, replica, errorMessage);
+ errors.add(new StreamError(this.tokenRange, replica,
errorMessage));
clean(replica, sessionID);
return false;
}
}
- private void sendSSTableToOneReplica(Path dataFile,
+ private void sendSSTableToOneReplica(SortedSSTableWriter.PreparedSSTable
preparedSSTable,
int ssTableIdx,
- RingInstance instance,
- Map<Path, Digest> fileHashes) throws
IOException
+ RingInstance instance) throws
IOException
{
- try (DirectoryStream<Path> componentFileStream =
Files.newDirectoryStream(dataFile.getParent(),
-
SSTables.getSSTableBaseName(dataFile) + "*"))
+ for (Path componentFile : preparedSSTable.files())
{
- for (Path componentFile : componentFileStream)
+ // send data component the last
+ if (preparedSSTable.dataFile().equals(componentFile))
Review Comment:
Theoretically it could, but only if data file was not generated during
sstable creation (e.g. `remainingSSTablesAfterClose` contained some sstable
files, but without `Data.db`). I have replaced the operands in equality check,
and added preconditions check.
--
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]
