@SuppressWarnings("unchecked") @Test public void testPublishingMessageAckModeNoUnitOfWorkButKafkaReturnTimeoutOnWriteShouldBeMarkedAsFailed() { String testTopic = "testSendMessagesAckNoUnitOfWorkWithTimeout"; testProducerFactory = mock(DefaultProducerFactory.class); when(testProducerFactory.confirmationMode()).thenReturn(ConfirmationMode.WAIT_FOR_ACK); Producer<String, byte[]> testProducer = mock(Producer.class); when(testProducerFactory.createProducer()).thenReturn(testProducer); Future<RecordMetadata> timeoutFuture1 = new CompletableFuture<>(); Future<RecordMetadata> timeoutFuture2 = new CompletableFuture<>(); when(testProducer.send(any())).thenReturn(timeoutFuture1).thenReturn(timeoutFuture2); testSubject = buildPublisher(testTopic); List<GenericDomainEventMessage<String>> testMessages = domainMessages("98765", 2); assertThatExceptionOfType(EventPublicationFailedException.class) .isThrownBy(() -> eventBus.publish(testMessages)); assertThat(monitor.successCount()).isZero(); assertThat(monitor.failureCount()).isEqualTo(1); assertThat(monitor.ignoreCount()).isZero(); }
public <T extends EventMessage<?>> void send(T event) { UnitOfWork<?> uow = CurrentUnitOfWork.get(); MonitorCallback monitorCallback = messageMonitor.onMessageIngested(event); Producer<K, V> producer = producerFactory.createProducer(); ConfirmationMode confirmationMode = producerFactory.confirmationMode(); if (confirmationMode.isTransactional()) { tryBeginTxn(producer); } Future<RecordMetadata> publishStatus = producer.send(messageConverter.createKafkaMessage(event, topic)); uow.onPrepareCommit(u -> { if (confirmationMode.isTransactional()) { tryCommit(producer, monitorCallback); } else if (confirmationMode.isWaitForAck()) { waitForPublishAck(publishStatus, monitorCallback); } tryClose(producer); }); uow.onRollback(u -> { if (confirmationMode.isTransactional()) { tryRollback(producer); } tryClose(producer); }); }
KafkaPublisher { public <T extends EventMessage<?>> void send(T event) { UnitOfWork<?> uow = CurrentUnitOfWork.get(); MonitorCallback monitorCallback = messageMonitor.onMessageIngested(event); Producer<K, V> producer = producerFactory.createProducer(); ConfirmationMode confirmationMode = producerFactory.confirmationMode(); if (confirmationMode.isTransactional()) { tryBeginTxn(producer); } Future<RecordMetadata> publishStatus = producer.send(messageConverter.createKafkaMessage(event, topic)); uow.onPrepareCommit(u -> { if (confirmationMode.isTransactional()) { tryCommit(producer, monitorCallback); } else if (confirmationMode.isWaitForAck()) { waitForPublishAck(publishStatus, monitorCallback); } tryClose(producer); }); uow.onRollback(u -> { if (confirmationMode.isTransactional()) { tryRollback(producer); } tryClose(producer); }); } }
KafkaPublisher { public <T extends EventMessage<?>> void send(T event) { UnitOfWork<?> uow = CurrentUnitOfWork.get(); MonitorCallback monitorCallback = messageMonitor.onMessageIngested(event); Producer<K, V> producer = producerFactory.createProducer(); ConfirmationMode confirmationMode = producerFactory.confirmationMode(); if (confirmationMode.isTransactional()) { tryBeginTxn(producer); } Future<RecordMetadata> publishStatus = producer.send(messageConverter.createKafkaMessage(event, topic)); uow.onPrepareCommit(u -> { if (confirmationMode.isTransactional()) { tryCommit(producer, monitorCallback); } else if (confirmationMode.isWaitForAck()) { waitForPublishAck(publishStatus, monitorCallback); } tryClose(producer); }); uow.onRollback(u -> { if (confirmationMode.isTransactional()) { tryRollback(producer); } tryClose(producer); }); } protected  KafkaPublisher(Builder<K, V> builder); }
KafkaPublisher { public <T extends EventMessage<?>> void send(T event) { UnitOfWork<?> uow = CurrentUnitOfWork.get(); MonitorCallback monitorCallback = messageMonitor.onMessageIngested(event); Producer<K, V> producer = producerFactory.createProducer(); ConfirmationMode confirmationMode = producerFactory.confirmationMode(); if (confirmationMode.isTransactional()) { tryBeginTxn(producer); } Future<RecordMetadata> publishStatus = producer.send(messageConverter.createKafkaMessage(event, topic)); uow.onPrepareCommit(u -> { if (confirmationMode.isTransactional()) { tryCommit(producer, monitorCallback); } else if (confirmationMode.isWaitForAck()) { waitForPublishAck(publishStatus, monitorCallback); } tryClose(producer); }); uow.onRollback(u -> { if (confirmationMode.isTransactional()) { tryRollback(producer); } tryClose(producer); }); } protected  KafkaPublisher(Builder<K, V> builder); static Builder<K, V> builder(); void send(T event); void shutDown(); }
KafkaPublisher { public <T extends EventMessage<?>> void send(T event) { UnitOfWork<?> uow = CurrentUnitOfWork.get(); MonitorCallback monitorCallback = messageMonitor.onMessageIngested(event); Producer<K, V> producer = producerFactory.createProducer(); ConfirmationMode confirmationMode = producerFactory.confirmationMode(); if (confirmationMode.isTransactional()) { tryBeginTxn(producer); } Future<RecordMetadata> publishStatus = producer.send(messageConverter.createKafkaMessage(event, topic)); uow.onPrepareCommit(u -> { if (confirmationMode.isTransactional()) { tryCommit(producer, monitorCallback); } else if (confirmationMode.isWaitForAck()) { waitForPublishAck(publishStatus, monitorCallback); } tryClose(producer); }); uow.onRollback(u -> { if (confirmationMode.isTransactional()) { tryRollback(producer); } tryClose(producer); }); } protected  KafkaPublisher(Builder<K, V> builder); static Builder<K, V> builder(); void send(T event); void shutDown();  }