### Question:
KafkaEventSender implements EventSender { @Override public <K extends Entity.Key<S>, S extends Specification> CompletableFuture<Void> send(@NonNull Event<K, S> event) { var avroEvent = converter.toAvro(event); return send(avroEvent.getKey(), avroEvent.getValue()); } KafkaEventSender(Config config, EventCorrelator correlator);  KafkaEventSender(Config config); @Override CompletableFuture<Void> send(@NonNull Event<K, S> event); @Override void close();  }### Answer:
@Test public void nullCorrelatorSuccess() { var correlationStrategy = new NullCorrelationStrategy(); var underTest = new KafkaEventSender(config, correlationStrategy, converter, producer); var result = underTest.send(event); verify(producer).send(recordCaptor.capture(), callbackCaptor.capture()); var record = recordCaptor.getValue(); assertThat(record.topic(), is("topic")); assertThat(record.key(), is(avroKey)); assertThat(record.value(), is(avroValue)); assertThat(record.headers().toArray().length, is(0)); var callback = callbackCaptor.getValue(); assertThat(result.isDone(), is(false)); var recordMetadata = mock(RecordMetadata.class); callback.onCompletion(recordMetadata, null); assertThat(result.isDone(), is(true)); }
@Test public void correlatorSuccess() { var correlator = mock(EventCorrelator.class); var correlationStrategy = new CorrelationStrategyImpl(correlator); var underTest = new KafkaEventSender(config, correlationStrategy, converter, producer); when(correlator.register(any())).thenReturn("correlationId"); var result = underTest.send(event); verify(correlator).register(result); verify(producer).send(recordCaptor.capture(), callbackCaptor.capture()); var record = recordCaptor.getValue(); assertThat(record.topic(), is("topic")); assertThat(record.key(), is(avroKey)); assertThat(record.value(), is(avroValue)); assertThat(record.headers().toArray().length, is(1)); var callback = callbackCaptor.getValue(); assertThat(result.isDone(), is(false)); var recordMetadata = mock(RecordMetadata.class); callback.onCompletion(recordMetadata, null); assertThat(result.isDone(), is(false)); }
@Test public void nullCorrelatorFailure() { var correlationStrategy = new NullCorrelationStrategy(); var underTest = new KafkaEventSender(config, correlationStrategy, converter, producer); var result = underTest.send(event); verify(producer).send(recordCaptor.capture(), callbackCaptor.capture()); var callback = callbackCaptor.getValue(); assertThat(result.isDone(), is(false)); var e = new Exception(); callback.onCompletion(null, e); assertThat(result.isCompletedExceptionally(), is(true)); }
@Test public void correlatorFailure() { var correlator = mock(EventCorrelator.class); var correlationStrategy = new CorrelationStrategyImpl(correlator); var underTest = new KafkaEventSender(config, correlationStrategy, converter, producer); when(correlator.register(any())).thenReturn("correlationId"); var result = underTest.send(event); verify(correlator).register(any()); verify(producer).send(recordCaptor.capture(), callbackCaptor.capture()); var record = recordCaptor.getValue(); assertThat(record.topic(), is("topic")); assertThat(record.key(), is(avroKey)); assertThat(record.value(), is(avroValue)); assertThat(record.headers().toArray().length, is(1)); var callback = callbackCaptor.getValue(); assertThat(result.isDone(), is(false)); var e = new Exception(); callback.onCompletion(null, e); assertThat(result.isCompletedExceptionally(), is(false)); verify(correlator).failed("correlationId", e); }