yifan-c commented on code in PR #239:
URL:
https://github.com/apache/cassandra-analytics/pull/239#discussion_r3909469673
##########
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);
Review Comment:
`String.join("\n", stackTrace)` is equivalent to just `stackTrace`, right?
The stacktrace are already aggregated in the statement above.
##########
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:
Can `dataFile()` return null?
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSession.java:
##########
Review Comment:
The effect of this method has no use. It can be deleted, along with
`isFileStreamed`
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSession.java:
##########
@@ -85,20 +82,19 @@ protected void onSSTablesProduced(Set<SSTableDescriptor>
sstables)
// 2. validate the sstables
// 3. send the sstables to all replicas
// 4. remove the sstables once sent
- Map<Path, Digest> fileDigests =
sstableWriter.prepareSStablesToSend(writerContext, sstables);
+ SortedSSTableWriter.PreparedSSTables preparedSSTables =
sstableWriter.prepareSStablesToSend(writerContext, sstables);
// retain only the SSTable data components
Review Comment:
Please remove this comment, it is stale with the removal of `.filter`
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriter.java:
##########
@@ -438,4 +437,107 @@ public Map<Path, Digest> fileDigestMap()
{
return Collections.unmodifiableMap(overallFileDigests);
}
+
+ public PreparedSSTables remainingSSTablesAfterClose()
+ {
+ return remainingSSTablesAfterClose;
+ }
+
+ /**
+ * Helper class representing list of newly generated sstables.
+ */
+ public static class PreparedSSTables
+ {
+ private static final PreparedSSTables EMPTY = new PreparedSSTables();
+
+ private final Map<String, PreparedSSTable> sstables = new HashMap<>();
// indexed by base file name
+
+ public PreparedSSTable addIfAbsent(Path path)
+ {
+ String baseName = SSTables.getSSTableDescriptor(path).baseFilename;
+ return sstables.computeIfAbsent(baseName, (__) -> new
PreparedSSTable());
Review Comment:
Based on the impl, `getOrPrepareSSTable` method name fits better.
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/SortedSSTableWriter.java:
##########
@@ -438,4 +437,107 @@ public Map<Path, Digest> fileDigestMap()
{
return Collections.unmodifiableMap(overallFileDigests);
}
+
+ public PreparedSSTables remainingSSTablesAfterClose()
+ {
+ return remainingSSTablesAfterClose;
+ }
+
+ /**
+ * Helper class representing list of newly generated sstables.
+ */
+ public static class PreparedSSTables
+ {
+ private static final PreparedSSTables EMPTY = new PreparedSSTables();
Review Comment:
`EMPTY` is mutable. Caller can call `addIfAbsent` on it and mess up the
state.
##########
cassandra-analytics-core/src/main/java/org/apache/cassandra/spark/bulkwriter/DirectStreamSession.java:
##########
@@ -153,22 +149,15 @@ protected StreamResult doFinalizeStream()
@Override
protected void sendRemainingSSTables()
{
- try (DirectoryStream<Path> dataFileStream =
Files.newDirectoryStream(sstableWriter.getOutDir(), "*Data.db"))
+ try
{
- for (Path dataFile : dataFileStream)
- {
- if (isFileStreamed(dataFile))
Review Comment:
`StreamSession.isFileStreamed` is now dead. Can you remove the code in the
base class?
--
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]
