simplify KafkaSender by removing Optional wrapper

this class is only ever constructed with a non-empty Optional, so we can simplify it by removing the Optional.
spotify · Dec 7, 2016 · fbe6378 · fbe6378
1 parent f704368
commit fbe6378
Show file tree

Hide file tree

Showing 2 changed files with 7 additions and 15 deletions.
diff --git a/helios-services/src/main/java/com/spotify/helios/servicescommon/EventSenderFactory.java b/helios-services/src/main/java/com/spotify/helios/servicescommon/EventSenderFactory.java
@@ -52,9 +52,7 @@ public static List<EventSender> build(
     final Optional<KafkaProducer<String, byte[]>> kafkaProducer =
         kafkaClientProvider.getDefaultProducer();
 
-    if (kafkaProducer.isPresent()) {
-      senders.add(new KafkaSender(kafkaProducer));
-    }
+    kafkaProducer.ifPresent(producer -> senders.add(new KafkaSender(producer)));
 
     final LifecycleEnvironment lifecycle = environment.lifecycle();
 

diff --git a/helios-services/src/main/java/com/spotify/helios/servicescommon/KafkaSender.java b/helios-services/src/main/java/com/spotify/helios/servicescommon/KafkaSender.java
@@ -24,18 +24,16 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.Optional;
-
 /**
  * A class that wraps {@link org.apache.kafka.clients.producer.KafkaProducer}.
  */
 public class KafkaSender implements EventSender {
 
   private static final Logger log = LoggerFactory.getLogger(KafkaSender.class);
 
-  private final Optional<KafkaProducer<String, byte[]>> kafkaProducer;
+  private final KafkaProducer<String, byte[]> kafkaProducer;
 
-  public KafkaSender(final Optional<KafkaProducer<String, byte[]>> kafkaProducer) {
+  public KafkaSender(final KafkaProducer<String, byte[]> kafkaProducer) {
     this.kafkaProducer = kafkaProducer;
   }
 
@@ -46,18 +44,14 @@ public void start() throws Exception {
 
   @Override
   public void stop() throws Exception {
-    kafkaProducer.ifPresent(KafkaProducer::close);
+    kafkaProducer.close();
   }
 
   private void send(final KafkaRecord kafkaRecord) {
-    if (kafkaProducer.isPresent()) {
-      final ProducerRecord<String, byte[]> record =
-          new ProducerRecord<>(kafkaRecord.getKafkaTopic(), kafkaRecord.getKafkaData());
+    final ProducerRecord<String, byte[]> record =
+        new ProducerRecord<>(kafkaRecord.getKafkaTopic(), kafkaRecord.getKafkaData());
 
-      kafkaProducer.get().send(record, new LoggingCallback());
-    } else {
-      log.debug("KafkaProducer isn't set. Not sending anything.");
-    }
+    kafkaProducer.send(record, new LoggingCallback());
   }
 
   @Override