PublisherHandler implements EventConsumer { @Override public void accept(SequenceEvent sequenceEvent, RunState state) { final WorkflowInstance workflowInstance = state.workflowInstance(); switch (state.state()) { case SUBMITTED: try { Preconditions.checkArgument(state.data().executionDescription().isPresent()); final ExecutionDescription executionDescription = state.data().executionDescription().get(); retryer.call( meteredPublishing(() -> publisher.deploying(workflowInstance, executionDescription), stats, DEPLOYING, SUBMITTED.name())); } catch (Exception e) { stats.recordPublishingError(DEPLOYING, SUBMITTED.name()); LOG.error("Failed to publish event for {} state", SUBMITTED.name(), e); } break; case RUNNING: final String type = "deployed"; try { Preconditions.checkArgument(state.data().executionDescription().isPresent()); final ExecutionDescription executionDescription = state.data().executionDescription().get(); retryer.call( meteredPublishing(() -> publisher.deployed(workflowInstance, executionDescription), stats, type, RUNNING.name())); } catch (Exception e) { stats.recordPublishingError(DEPLOYED, RUNNING.name()); LOG.error("Failed to publish event for {} state", RUNNING.name(), e); } break; default: } } PublisherHandler(Publisher publisher, Stats stats); @VisibleForTesting  PublisherHandler(Publisher publisher, Stats stats, Retryer<Void> retryer); @Override void accept(SequenceEvent sequenceEvent, RunState state);  }
@Test public void testPublishesRollingOutStateOnSubmitted() throws Exception { ExecutionDescription executionDescription = ExecutionDescription.builder() .dockerImage(DOCKER_IMAGE) .commitSha(COMMIT_SHA) .build(); RunState runState = RunState.create( WORKFLOW_INSTANCE, RunState.State.SUBMITTED, StateData.newBuilder() .executionId("exec1") .executionDescription(executionDescription) .build()); publisherHandler.accept(sequenceEvent, runState); verify(publisher).deploying(WORKFLOW_INSTANCE, executionDescription); verify(stats).recordPublishing("deploying", "SUBMITTED"); }
@Test public void testPublishesDoneStateOnRunning() throws Exception { ExecutionDescription executionDescription = ExecutionDescription.builder() .dockerImage(DOCKER_IMAGE) .commitSha(COMMIT_SHA) .build(); RunState runState = RunState.create( WORKFLOW_INSTANCE, RunState.State.RUNNING, StateData.newBuilder() .executionId("exec1") .executionDescription(executionDescription) .build()); publisherHandler.accept(sequenceEvent, runState); verify(publisher).deployed(WORKFLOW_INSTANCE, executionDescription); verify(stats).recordPublishing("deployed", "RUNNING"); }
@Test public void shouldRetryPublishesOnSubmitted() throws Exception { publisherHandler = new PublisherHandler(new FailingPublisher(publisher, 2), stats); ExecutionDescription executionDescription = ExecutionDescription.builder() .dockerImage(DOCKER_IMAGE) .commitSha(COMMIT_SHA) .build(); RunState runState = RunState.create( WORKFLOW_INSTANCE, RunState.State.SUBMITTED, StateData.newBuilder() .executionId("exec1") .executionDescription(executionDescription) .build()); publisherHandler.accept(sequenceEvent, runState); verify(publisher).deploying(WORKFLOW_INSTANCE, executionDescription); verify(stats, times(2)).recordPublishingError("deploying", "SUBMITTED"); verify(stats).recordPublishing("deploying", "SUBMITTED"); }
@Test public void shouldFailEventuallyOnSubmitted() throws Exception { doThrow(new IOException()).when(publisher).deploying(any(), any()); publisherHandler = new PublisherHandler( publisher, stats, NO_RETRIES); ExecutionDescription executionDescription = ExecutionDescription.builder() .dockerImage(DOCKER_IMAGE) .commitSha(COMMIT_SHA) .build(); RunState runState = RunState.create( WORKFLOW_INSTANCE, RunState.State.SUBMITTED, StateData.newBuilder() .executionId("exec1") .executionDescription(executionDescription) .build()); publisherHandler.accept(sequenceEvent, runState); verify(publisher).deploying(WORKFLOW_INSTANCE, executionDescription); verify(stats, times(2)).recordPublishingError("deploying", "SUBMITTED"); }
@Test public void shouldRetryPublishesOnRunning() throws Exception { publisherHandler = new PublisherHandler(new FailingPublisher(publisher, 2), stats); ExecutionDescription executionDescription = ExecutionDescription.builder() .dockerImage(DOCKER_IMAGE) .commitSha(COMMIT_SHA) .build(); RunState runState = RunState.create( WORKFLOW_INSTANCE, RunState.State.RUNNING, StateData.newBuilder() .executionId("exec1") .executionDescription(executionDescription) .build()); publisherHandler.accept(sequenceEvent, runState); verify(publisher).deployed(WORKFLOW_INSTANCE, executionDescription); verify(stats, times(2)).recordPublishingError("deployed", "RUNNING"); verify(stats).recordPublishing("deployed", "RUNNING"); }
@Test public void shouldFailEventuallyOnRunning() throws Exception { doThrow(new IOException()).when(publisher).deployed(any(), any()); publisherHandler = new PublisherHandler(publisher, stats, NO_RETRIES); ExecutionDescription executionDescription = ExecutionDescription.builder() .dockerImage(DOCKER_IMAGE) .commitSha(COMMIT_SHA) .build(); RunState runState = RunState.create( WORKFLOW_INSTANCE, RunState.State.RUNNING, StateData.newBuilder() .executionId("exec1") .executionDescription(executionDescription) .build()); publisherHandler.accept(sequenceEvent, runState); verify(publisher).deployed(WORKFLOW_INSTANCE, executionDescription); verify(stats, times(2)).recordPublishingError("deployed", "RUNNING"); }