|
1 | 1 | /* |
2 | | - * Copyright 2009-2023 the original author or authors. |
| 2 | + * Copyright 2009-2024 the original author or authors. |
3 | 3 | * |
4 | 4 | * Licensed under the Apache License, Version 2.0 (the "License"); |
5 | 5 | * you may not use this file except in compliance with the License. |
@@ -251,24 +251,22 @@ protected Set<StepExecution> doHandle(StepExecution managerStepExecution, |
251 | 251 |
|
252 | 252 | private Set<StepExecution> pollReplies(final StepExecution managerStepExecution, final Set<StepExecution> split) |
253 | 253 | throws Exception { |
254 | | - final Set<StepExecution> result = new HashSet<>(split.size()); |
| 254 | + Set<Long> partitionStepExecutionIds = split.stream().map(StepExecution::getId).collect(Collectors.toSet()); |
255 | 255 |
|
256 | 256 | Callable<Set<StepExecution>> callback = () -> { |
257 | | - Set<Long> currentStepExecutionIds = split.stream().map(StepExecution::getId).collect(Collectors.toSet()); |
258 | 257 | JobExecution jobExecution = jobExplorer.getJobExecution(managerStepExecution.getJobExecutionId()); |
259 | | - jobExecution.getStepExecutions() |
| 258 | + Set<StepExecution> finishedStepExecutions = jobExecution.getStepExecutions() |
260 | 259 | .stream() |
261 | | - .filter(stepExecution -> currentStepExecutionIds.contains(stepExecution.getId())) |
262 | | - .filter(stepExecution -> !result.contains(stepExecution)) |
| 260 | + .filter(stepExecution -> partitionStepExecutionIds.contains(stepExecution.getId())) |
263 | 261 | .filter(stepExecution -> !stepExecution.getStatus().isRunning()) |
264 | | - .forEach(result::add); |
| 262 | + .collect(Collectors.toSet()); |
265 | 263 |
|
266 | 264 | if (logger.isDebugEnabled()) { |
267 | 265 | logger.debug(String.format("Currently waiting on %s partitions to finish", split.size())); |
268 | 266 | } |
269 | 267 |
|
270 | | - if (result.size() == split.size()) { |
271 | | - return result; |
| 268 | + if (finishedStepExecutions.size() == split.size()) { |
| 269 | + return finishedStepExecutions; |
272 | 270 | } |
273 | 271 | else { |
274 | 272 | return null; |
|
0 commit comments