LakshSingla commented on code in PR #13062:
URL: https://github.com/apache/druid/pull/13062#discussion_r974235779
##########
extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerImpl.java:
##########
@@ -856,7 +875,33 @@ public void onSuccess(final List<Object>
workerResultAndOutputChannelsResolved)
for (OutputChannel channel : outputChannels.getAllChannels()) {
stageOutputs.computeIfAbsent(stageDef.getId(), ignored1 -> new
ConcurrentHashMap<>())
.computeIfAbsent(channel.getPartitionNumber(),
ignored2 -> channel.getReadableChannel());
+
}
+
+ if (durableStageStorageEnabled) {
+ // Once the outputs channels have been resolved and are ready
for reading, the worker appends the filename
+ // with a special marker flag and adds it to the
+ DurableStorageOutputChannelFactory
durableStorageOutputChannelFactory =
+
DurableStorageOutputChannelFactory.createStandardImplementation(
+ task.getControllerTaskId(),
+ task().getWorkerNumber(),
+ stageDef.getStageNumber(),
+ task().getId(),
+ frameContext.memoryParameters().getStandardFrameSize(),
+ MSQTasks.makeStorageConnector(context.injector())
+ );
+ try {
+
durableStorageOutputChannelFactory.createSuccessFile(task.getId());
Review Comment:
Another place where I wanted to check is if a race condition can happen if
multiple workers are simultaneously executing this piece of code. Is there any
guarantee in the storage connector that ensures that the final file will
contain the input from a single worker? I have added a check in the
`createSuccessFile` that checks if the file is already created or not for
starters. If the write operation for the connector is atomic, I suppose that we
should be fine.
--
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]
