Merge pull request #164 from splunk/channel-expiration-logging

Channel expiration logging - improvement/bugfix

Author: Donald Tregonning

src/main/java/com/splunk/hecclient/HecAckPoller.java

@@ -44,6 +44,7 @@ public final class HecAckPoller implements Poller {
     private ScheduledThreadPoolExecutor scheduler;
     private ExecutorService executorService;
     private AtomicBoolean started;
+    private AtomicBoolean stickySessionStarted;
 
     public HecAckPoller(PollerCallback cb) {
         outstandingEventBatches = new ConcurrentHashMap<>();
@@ -53,6 +54,11 @@ public HecAckPoller(PollerCallback cb) {
         pollThreads = 2;
         pollerCallback = cb;
         started = new AtomicBoolean(false);
+        stickySessionStarted = new AtomicBoolean(false);
+    }
+
+    public void setStickySessionToTrue() {
+        stickySessionStarted.compareAndSet(false, true);
     }
 
     @Override
@@ -201,6 +207,9 @@ public int getAckPollInterval() {
      * @since        1.1.0
      */
     public void stickySessionHandler(HecChannel channel) {
+        if (!stickySessionStarted.get()) {
+            return;
+        }
         String oldChannelId = channel.getId();
         channel.setAvailable(false);
         log.info("Channel {} set to be not available", oldChannelId);
@@ -228,6 +237,7 @@ public void stickySessionHandler(HecChannel channel) {
 
         channel.setAvailable(true);
         log.info("Channel {} is available", newChannelId);
+        stickySessionStarted.compareAndSet(true, false);
     }
 
     private void poll() {
@@ -305,6 +315,7 @@ private void handleAckPollResponse(String resp, HecChannel channel) {
             log.error("failed to handle ack polled result", ex);
             return;
         }
+        stickySessionHandler(channel);
         handleAckPollResult(channel, ackPollResult);
     }
 

src/main/java/com/splunk/hecclient/Indexer.java

@@ -124,6 +124,8 @@ public boolean send(final EventBatch batch) {
             return false;
         }
 
+        poller.stickySessionHandler(channel);
+
         // we are all good
         poller.add(this.channel, batch, resp);
         log.debug("sent {} events to splunk through channel={} indexer={}", batch.size(), channel.getId(), getBaseUrl());
@@ -166,7 +168,7 @@ private String readAndCloseResponse(CloseableHttpResponse resp) {
 
         if((resp.getHeaders("Set-Cookie") != null) && (resp.getHeaders("Set-Cookie").length > 0)) {
             log.info("Sticky session expiry detected, will cleanup old channel and its associated batches");
-            poller.stickySessionHandler(channel);
+            poller.setStickySessionToTrue();
         }
 
         int status = resp.getStatusLine().getStatusCode();

src/main/java/com/splunk/hecclient/Poller.java

@@ -21,7 +21,8 @@ public interface Poller {
     void add(HecChannel channel, EventBatch batch, String response);
     void fail(HecChannel channel, EventBatch batch, Exception ex);
     void stickySessionHandler(HecChannel channel);
-    // minimum load channel
+    void setStickySessionToTrue();
+        // minimum load channel
     HecChannel getMinLoadChannel();
     long getTotalOutstandingEventBatches();
 }

src/main/java/com/splunk/hecclient/ResponsePoller.java

@@ -79,4 +79,6 @@ public void add(HecChannel channel, EventBatch batch, String resp) {
             callback.onEventCommitted(Arrays.asList(batch));
         }
     }
+
+    public void setStickySessionToTrue() {}
 }

src/test/java/com/splunk/hecclient/HecAckPollerTest.java

@@ -347,6 +347,7 @@ public void stickySessionHandler() {
         UnitUtil.milliSleep(3000);
 
         String oldId = ch.getId();
+        poller.setStickySessionToTrue();
         poller.stickySessionHandler(ch);
         Assert.assertNotEquals(oldId, ch.getId());
 

src/test/java/com/splunk/hecclient/PollerMock.java

@@ -88,4 +88,6 @@ public Exception getException() {
     public String getResponse() {
         return response;
     }
+    public void setStickySessionToTrue() {
+    }
 }