@Override
    public ConcurrentJobModificationResolveResult resolve(Job localJob, Job storageProviderJob) {
        //why: we use the JobVersioner to bump the version so it matched the one from the DB
        new JobVersioner(localJob);
        storageProvider.save(localJob);
        return ConcurrentJobModificationResolveResult.succeeded(localJob);
    }
@Test
    void ifJobHasEnqueuedStateAndWasScheduledTooEarlyByJobZooKeeperItWillResolveLocalJobAndSaveIt() {
        final Job scheduledJob = aJob()
                .withFailedState()
                .withScheduledState()
                .withVersion(5)
                .build();
        final Job enqueuedJob = aCopyOf(scheduledJob)
                .withEnqueuedState(Instant.now())
                .withVersion(4)
                .build();


        final ConcurrentJobModificationResolveResult resolveResult = allowedStateChange.resolve(enqueuedJob, scheduledJob);

        verify(storageProvider).save(jobCaptor.capture());

        assertThat(jobCaptor.getValue())
                .hasVersion(5)
                .hasStates(StateName.FAILED, StateName.SCHEDULED, StateName.ENQUEUED);
        assertThat(resolveResult.failed()).isFalse();
        assertThat(resolveResult.getLocalJob()).isEqualTo(enqueuedJob);
    }