From 481937e5bfb1defc22b172ac6b19ae6614c36e1b Mon Sep 17 00:00:00 2001
From: acabezas <acabezas@homeaway.com>
Date: Thu, 27 Jun 2019 22:12:44 -0500
Subject: [PATCH] Fix small bug with Consumers singleton.

---
 .../photon/client/consumer/Consumers.java      | 18 +++++++++++++++---
 1 file changed, 15 insertions(+), 3 deletions(-)

diff --git a/src/main/java/com/homeaway/datatools/photon/client/consumer/Consumers.java b/src/main/java/com/homeaway/datatools/photon/client/consumer/Consumers.java
index 340392f..01ac49c 100644
--- a/src/main/java/com/homeaway/datatools/photon/client/consumer/Consumers.java
+++ b/src/main/java/com/homeaway/datatools/photon/client/consumer/Consumers.java
@@ -55,12 +55,11 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Properties;
-import java.util.concurrent.ConcurrentMap;
 
 public class Consumers {
 
     private final PhotonDriver driver;
-    private final ConcurrentMap<ConsumerType, ConsumerFactory> factories = Maps.newConcurrentMap();
+    private ConsumerFactory consumerFactory;
     private BeamReaderLockManager beamReaderLockManager;
     private BeamReaderConfigManager beamReaderConfigManager;
     private PhotonScheduler beamReaderScheduler;
@@ -88,6 +87,14 @@ private static Consumers getInstance(final Properties properties) {
     }
 
     private ConsumerFactory getFactory(Properties properties) {
+        return Optional.ofNullable(consumerFactory)
+                .orElseGet(() -> {
+                    consumerFactory = buildFactory(properties);
+                    return consumerFactory;
+                });
+    }
+
+    private ConsumerFactory buildFactory(Properties properties) {
         try {
             PhotonDeserializer photonDeserializer = (PhotonDeserializer) Class.forName(properties.getProperty(PHOTON_DESERIALIZER_CLASS))
                     .getConstructor()
@@ -149,7 +156,12 @@ public static ConsumerFactory newConsumerFactory(final Properties properties) {
     }
 
     public static void shutDown() {
-        instance.driver.shutDown();
+        instance.shutDownFactory();
+    }
+
+    private void shutDownFactory() {
+        driver.shutDown();
+        consumerFactory = null;
     }
 
     enum ConsumerType {