KafkaBrokerMessageProducer implements BrokerMessageProducer { @Override public void send(Message message, CachedTopic cachedTopic, final PublishingCallback callback) { ProducerRecord<byte[], byte[]> producerRecord = createProducerRecord(message, cachedTopic); try { producers.get(cachedTopic.getTopic()).send(producerRecord, new SendCallback(message, cachedTopic.getTopic(), callback)); } catch (Exception e) { callback.onUnpublished(message, cachedTopic.getTopic(), e); } } @Inject  KafkaBrokerMessageProducer(Producers producers,
                                      HermesMetrics metrics,
                                      KafkaHeaderFactory kafkaHeaderFactory,
                                      ConfigFactory configFactory); @Override void send(Message message, CachedTopic cachedTopic, final PublishingCallback callback); @Override boolean isTopicAvailable(CachedTopic cachedTopic);  }
@Test public void shouldPublishOnTopicUsingKafkaTopicName() throws InterruptedException { producer.send(MESSAGE, cachedTopic, new DoNothing()); List<ProducerRecord<byte[], byte[]>> records = leaderConfirmsProducer.history(); assertThat(records.size()).isEqualTo(1); assertThat(records.get(0).topic()).isEqualTo("ns_group.topic"); }
@Test public void shouldCallCallbackOnSend() throws InterruptedException { final AtomicBoolean callbackCalled = new AtomicBoolean(false); producer.send(MESSAGE, cachedTopic, new PublishingCallback() { @Override public void onUnpublished(Message message, Topic topic, Exception exception) { callbackCalled.set(true); } @Override public void onPublished(Message message, Topic topic) { callbackCalled.set(true); } }); await().until(callbackCalled::get); }
@Test public void shouldUseEveryoneConfirmProducerForTopicWithAckAll() { Topic topic = topic("group.all").withAck(Topic.Ack.ALL).build(); CachedTopic cachedTopic = new CachedTopic(topic, hermesMetrics, kafkaNamesMapper.toKafkaTopics(topic)); producer.send(MESSAGE, cachedTopic, new DoNothing()); List<ProducerRecord<byte[], byte[]>> records = everyoneConfirmProducer.history(); assertThat(records.size()).isEqualTo(1); assertThat(records.get(0).topic()).isEqualTo("ns_group.all"); }