prateekm commented on code in PR #1682:
URL: https://github.com/apache/samza/pull/1682#discussion_r1297765935
##########
samza-core/src/main/scala/org/apache/samza/storage/ContainerStorageManagerRestoreUtil.java:
##########
@@ -170,31 +170,23 @@ private static CompletableFuture<Map<TaskName,
Checkpoint>> restoreAllTaskInstan
checkpointManager, taskRestoreManager, config,
taskInstanceMetrics, executor,
taskBackendFactoryToStoreNames.get(taskInstanceName).get(factoryName),
loggedStoreDir,
jobContext, containerModel);
- try {
- newTaskCheckpoints.put(taskInstanceName, future);
- } catch (Exception e) {
- String msg = String.format("DeletedException during restore
task: %s after retrying to get deleted blobs.", taskName);
- throw new SamzaException(msg, e);
- } finally {
+ future.whenComplete((r, e) -> {
Review Comment:
Actually @dxichen suggestion is even better, since this needs to happen
regardless of original or retried restore. Just chain this close operation to
the future that is currently being returned/added to the map at the end of the
operation, and return/add the resulting future instead.
--
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]
