in src/main/java/com/googlesource/gerrit/plugins/kafka/session/KafkaSession.java [110:130]
private boolean publishAsync(String topic, String messageBody) {
try {
Future<RecordMetadata> future =
producer.send(
new ProducerRecord<>(topic, Long.toString(System.nanoTime()), messageBody),
(metadata, e) -> {
if (metadata != null && e == null) {
LOGGER.debug("The offset of the record we just sent is: {}", metadata.offset());
publisherMetrics.incrementBrokerPublishedMessage();
} else {
LOGGER.error("Cannot send the message", e);
publisherMetrics.incrementBrokerFailedToPublishMessage();
}
});
return future != null;
} catch (Throwable e) {
LOGGER.error("Cannot send the message", e);
publisherMetrics.incrementBrokerFailedToPublishMessage();
return false;
}
}