Copilot commented on code in PR #2513:
URL:
https://github.com/apache/shardingsphere-elasticjob/pull/2513#discussion_r3413505495
##########
kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/executor/ElasticJobExecutor.java:
##########
@@ -133,8 +133,7 @@ private void execute(final JobConfiguration jobConfig,
final ShardingContexts sh
try {
process(jobConfig, shardingContexts, executionSource);
} finally {
- // TODO Consider increasing the status of job failure, and how to
handle the overall loop of job failure
- jobFacade.registerJobCompleted(shardingContexts);
+ jobFacade.registerJobCompleted(shardingContexts,
itemErrorMessages.keySet());
Review Comment:
The new `failedItems` propagation is not asserted by tests: current unit
tests were relaxed to `any()` for the second argument, so they won’t catch
regressions where failed items are not passed (or are passed incorrectly).
Adding assertions that the collection contains exactly the failed sharding
items (and is empty on success) would better validate the bug fix intent.
##########
kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/executor/facade/AbstractJobFacade.java:
##########
@@ -121,9 +122,24 @@ public void registerJobBegin(final ShardingContexts
shardingContexts) {
*/
@Override
public void registerJobCompleted(final ShardingContexts shardingContexts) {
- executionService.registerJobCompleted(shardingContexts);
+ registerJobCompleted(shardingContexts, Collections.emptySet());
+ }
+
+ /**
+ * Register job completed.
+ * Failed items retain their running node so that failover can be
triggered for them.
+ *
+ * @param shardingContexts sharding contexts
+ * @param failedItems sharding items that failed during execution
+ */
+ @Override
+ public void registerJobCompleted(final ShardingContexts shardingContexts,
final Collection<Integer> failedItems) {
+ executionService.registerJobCompleted(shardingContexts, failedItems);
if (configService.load(true).isFailover()) {
-
failoverService.updateFailoverComplete(shardingContexts.getShardingItemParameters().keySet());
+ Collection<Integer> succeededItems =
shardingContexts.getShardingItemParameters().keySet().stream()
+ .filter(item -> !failedItems.contains(item))
+ .collect(Collectors.toSet());
+ failoverService.updateFailoverComplete(succeededItems);
Review Comment:
`failedItems.contains(item)` is called repeatedly during succeeded-item
calculation. Normalizing `failedItems` to a `Set` once (and handling potential
null) avoids repeated linear scans when callers provide a List and prevents NPE.
##########
kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/sharding/ExecutionService.java:
##########
@@ -75,12 +76,25 @@ public void registerJobBegin(final ShardingContexts
shardingContexts) {
* @param shardingContexts sharding contexts
*/
public void registerJobCompleted(final ShardingContexts shardingContexts) {
+ registerJobCompleted(shardingContexts, Collections.emptyList());
+ }
+
+ /**
+ * Register job completed.
+ * Failed items retain their running node so that failover can be
triggered for them.
+ *
+ * @param shardingContexts sharding contexts
+ * @param failedItems sharding items that failed during execution
+ */
+ public void registerJobCompleted(final ShardingContexts shardingContexts,
final Collection<Integer> failedItems) {
JobRegistry.getInstance().setJobRunning(jobName, false);
if (!configService.load(true).isMonitorExecution()) {
return;
}
for (int each : shardingContexts.getShardingItemParameters().keySet())
{
-
jobNodeStorage.removeJobNodeIfExisted(ShardingNode.getRunningNode(each));
+ if (!failedItems.contains(each)) {
+
jobNodeStorage.removeJobNodeIfExisted(ShardingNode.getRunningNode(each));
+ }
Review Comment:
`failedItems.contains(each)` inside the loop can be O(n²) when callers pass
a non-Set collection, and will throw NPE if `failedItems` is null. Consider
normalizing to a non-null `Set` once before the loop for both safety and
performance.
##########
kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/executor/facade/JobFacade.java:
##########
@@ -65,6 +65,15 @@ public interface JobFacade {
*/
void registerJobCompleted(ShardingContexts shardingContexts);
+ /**
+ * Register job completed.
+ * Failed items retain their running node so that failover can be
triggered for them.
+ *
+ * @param shardingContexts sharding contexts
+ * @param failedItems sharding items that failed during execution
+ */
+ void registerJobCompleted(ShardingContexts shardingContexts,
Collection<Integer> failedItems);
Review Comment:
Adding a new abstract method to a public interface is a source/binary
breaking change for any external `JobFacade` implementations. If backwards
compatibility matters here, consider making this overload a `default` method
that delegates to the existing 1-arg `registerJobCompleted(ShardingContexts)`
so existing implementations keep working and only facades that need failed-item
handling must override it.
##########
kernel/src/test/java/org/apache/shardingsphere/elasticjob/kernel/executor/facade/ShardingJobFacadeTest.java:
##########
@@ -35,6 +35,8 @@
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
import java.util.Arrays;
Review Comment:
Static imports are currently split into two blocks (ArgumentMatchers vs the
rest), with `java.util.*` imports in between. This differs from the import
grouping used by other tests in this module (e.g., `ElasticJobExecutorTest`)
and may violate formatting/checkstyle expectations; please regroup static
imports into a single block after the non-static imports.
##########
kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/executor/ElasticJobExecutor.java:
##########
@@ -133,8 +133,7 @@ private void execute(final JobConfiguration jobConfig,
final ShardingContexts sh
try {
process(jobConfig, shardingContexts, executionSource);
} finally {
- // TODO Consider increasing the status of job failure, and how to
handle the overall loop of job failure
- jobFacade.registerJobCompleted(shardingContexts);
+ jobFacade.registerJobCompleted(shardingContexts,
itemErrorMessages.keySet());
if (itemErrorMessages.isEmpty()) {
Review Comment:
`registerJobCompleted(shardingContexts, failedItems)` can bypass subclasses
that override only the 1-arg `registerJobCompleted(ShardingContexts)` with
additional completion side effects. For example, `SingleShardingJobFacade`
overrides the 1-arg method to update `next-job-instance-ip`, but does not
override the new overload, so that logic will no longer run now that the
executor always calls the 2-arg overload. Ensure any `JobFacade`
implementations with completion hooks override the new overload (or refactor to
share the hook) so behavior is preserved.
##########
kernel/src/main/java/org/apache/shardingsphere/elasticjob/kernel/internal/sharding/ExecutionService.java:
##########
@@ -75,12 +76,25 @@ public void registerJobBegin(final ShardingContexts
shardingContexts) {
* @param shardingContexts sharding contexts
*/
public void registerJobCompleted(final ShardingContexts shardingContexts) {
+ registerJobCompleted(shardingContexts, Collections.emptyList());
+ }
+
+ /**
+ * Register job completed.
+ * Failed items retain their running node so that failover can be
triggered for them.
+ *
+ * @param shardingContexts sharding contexts
+ * @param failedItems sharding items that failed during execution
+ */
+ public void registerJobCompleted(final ShardingContexts shardingContexts,
final Collection<Integer> failedItems) {
JobRegistry.getInstance().setJobRunning(jobName, false);
if (!configService.load(true).isMonitorExecution()) {
return;
}
for (int each : shardingContexts.getShardingItemParameters().keySet())
{
-
jobNodeStorage.removeJobNodeIfExisted(ShardingNode.getRunningNode(each));
+ if (!failedItems.contains(each)) {
+
jobNodeStorage.removeJobNodeIfExisted(ShardingNode.getRunningNode(each));
Review Comment:
There are existing unit tests for
`ExecutionService.registerJobCompleted(...)`, but none cover the new overload
behavior where failed items should retain their `/running` node while succeeded
items are removed. Adding a test for the new overload would help prevent
regressions in this critical state-management logic.
--
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]
