@Test(expected=IllegalArgumentException.class) public void testSendWithoutTopic() { eventuateKafkaProducer.send(null, event.getId(), event.getEventData()); }
public CompletableFuture<?> send(String topic, String key, String body) { CompletableFuture<Object> result = new CompletableFuture<>(); producer.send(new ProducerRecord<>(topic, key, body), (metadata, exception) -> { if (exception == null) result.complete(metadata); else result.completeExceptionally(exception); }); return result; }
EventuateKafkaProducer { public CompletableFuture<?> send(String topic, String key, String body) { CompletableFuture<Object> result = new CompletableFuture<>(); producer.send(new ProducerRecord<>(topic, key, body), (metadata, exception) -> { if (exception == null) result.complete(metadata); else result.completeExceptionally(exception); }); return result; } }
EventuateKafkaProducer { public CompletableFuture<?> send(String topic, String key, String body) { CompletableFuture<Object> result = new CompletableFuture<>(); producer.send(new ProducerRecord<>(topic, key, body), (metadata, exception) -> { if (exception == null) result.complete(metadata); else result.completeExceptionally(exception); }); return result; } EventuateKafkaProducer(); }
EventuateKafkaProducer { public CompletableFuture<?> send(String topic, String key, String body) { CompletableFuture<Object> result = new CompletableFuture<>(); producer.send(new ProducerRecord<>(topic, key, body), (metadata, exception) -> { if (exception == null) result.complete(metadata); else result.completeExceptionally(exception); }); return result; } EventuateKafkaProducer(); void setProducer(Producer<String, String> producer); CompletableFuture<?> send(String topic, String key, String body); void close(); }
EventuateKafkaProducer { public CompletableFuture<?> send(String topic, String key, String body) { CompletableFuture<Object> result = new CompletableFuture<>(); producer.send(new ProducerRecord<>(topic, key, body), (metadata, exception) -> { if (exception == null) result.complete(metadata); else result.completeExceptionally(exception); }); return result; } EventuateKafkaProducer(); void setProducer(Producer<String, String> producer); CompletableFuture<?> send(String topic, String key, String body); void close();  }