@Override
    public ConcurrentJobModificationResolveResult resolve(Job localJob, Job storageProviderJob) {
        if (localJob.getState() == StateName.SUCCEEDED && storageProviderJob.getState() == StateName.SUCCEEDED) {
            throw shouldNotHappenException("Should not happen as matches filter should be filtering out this StateChangeFilter");
        } else if (localJob.getState() == StateName.PROCESSING && storageProviderJob.getState() == StateName.SUCCEEDED) {
            localJob.delete("Job has already succeeded in StorageProvider");
            final Thread threadProcessingJob = jobSteward.getThreadProcessingJob(localJob);
            if (threadProcessingJob != null) {
                threadProcessingJob.interrupt();
            }
        }
        return ConcurrentJobModificationResolveResult.succeeded(localJob);
    }
@Test
    void ifJobSucceededWhileInProgress() {
        final Job jobInProgress = aJobInProgress().build();
        final Job jobInProgressWithUpdate = aCopyOf(jobInProgress).withMetadata("extra", "metadata").build();
        final Job succeededJob = aCopyOf(jobInProgress).withSucceededState().build();

        Thread mockThread = mock(Thread.class);
        when(jobZooKeeper.getThreadProcessingJob(jobInProgressWithUpdate)).thenReturn(mockThread);

        final ConcurrentJobModificationResolveResult resolveResult = allowedStateChange.resolve(jobInProgressWithUpdate, succeededJob);
        assertThat(resolveResult.failed()).isFalse();
        verify(mockThread).interrupt();
    }