Merge branch 'develop' into issue-138/header-support

Donald Tregonning · web-flow · commit 8c709a3fa8cd · 2018-08-21T13:31:38.000-07:00
diff --git a/.gitignore b/.gitignore
@@ -1,3 +1,4 @@
+
 # Compiled class file
 *.class
 
@@ -27,4 +28,5 @@ target/*
 splunk-kafka-connect/
 pom.xml.versionsBackup
 .classpath
-.project
+.project
+*.iml
diff --git a/build.sh b/build.sh
@@ -1,7 +1,7 @@
 #!/bin/bash
 
 # variables
-kafkaversion=0.11.0.2
+kafkaversion=2.0.0
 builddir=/tmp/splunk-kafka-connect-build/splunk-kafka-connect
 
 githash=`git rev-parse --short HEAD 2>/dev/null | sed "s/\(.*\)/@\1/"` # get current git hash
diff --git a/pom.xml b/pom.xml
@@ -6,7 +6,7 @@
 
     <groupId>com.github.splunk.kafka.connect</groupId>
     <artifactId>splunk-kafka-connect</artifactId>
-    <version>v1.0.0</version>
+    <version>v1.1.0</version>
     <name>splunk-kafka-connect</name>
 
     <properties>
diff --git a/src/main/java/com/splunk/hecclient/Event.java b/src/main/java/com/splunk/hecclient/Event.java
@@ -19,6 +19,7 @@
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.annotation.JsonSerialize;
+import com.fasterxml.jackson.databind.util.StdDateFormat;
 import org.slf4j.*;
 
 import java.io.*;
@@ -43,7 +44,13 @@ public abstract class Event {
     static final String SOURCE = "source";
     static final String SOURCETYPE = "sourcetype";
 
-    static final ObjectMapper jsonMapper = new ObjectMapper();
+    static final ObjectMapper jsonMapper;
+    static {
+        jsonMapper = new ObjectMapper();
+        jsonMapper.registerModule(new com.splunk.kafka.connect.JacksonStructModule());
+        jsonMapper.setDateFormat(StdDateFormat.instance);
+    }
+
     protected static final Logger log = LoggerFactory.getLogger(Event.class);
 
     @JsonSerialize(using = DoubleSerializer.class)
diff --git a/src/main/java/com/splunk/hecclient/HecAckPoller.java b/src/main/java/com/splunk/hecclient/HecAckPoller.java
@@ -132,7 +132,7 @@ public void add(HecChannel channel, EventBatch batch, String response) {
         }
 
         if (channelEvents.get(resp.getAckId()) != null) {
-            log.error("ackId={} already exists for channel={} index={}", resp.getAckId(), channel, channel.getIndexer());
+            log.warn("ackId={} already exists for channel={} index={} data may be duplicated in Splunk", resp.getAckId(), channel, channel.getIndexer());
             return;
         }
 
@@ -184,6 +184,52 @@ public int getAckPollInterval() {
         return ackPollInterval;
     }
 
+    /**
+     * StickySessionHandler is used to reassign channel id and fail the batches for that HecChannel.
+     * Also, the HecChannel will be unavailable during this period.
+     * StickySessionHandler follows the following flow:
+     * 1) Set channel unavailable
+     * 2) Get batches for the channel
+     * 3) Remove batches for the channel from the poller
+     * 4) Remove batches from kafka record tracker to fail them and resend
+     * 5) Remove channel
+     * 6) Change channel id
+     * 7) Set channel available
+     *
+     * @param  channel  HecChannel is the channel for which id has tobe changed and batches have to be failed.
+     * @see          HecChannel
+     * @since        1.1.0
+     */
+    public void stickySessionHandler(HecChannel channel) {
+        String oldChannelId = channel.getId();
+        channel.setAvailable(false);
+        log.info("Channel {} set to be not available", oldChannelId);
+        ConcurrentHashMap<Long, EventBatch> channelBatches = outstandingEventBatches.get(channel);
+        if(channelBatches != null && channelBatches.size() > 0) {
+            log.info("Failing {} batches for the channel {}, these will be resent by the connector.", channelBatches.size(), oldChannelId);
+            if (pollerCallback != null) {
+                List<EventBatch> expired = new ArrayList<>();
+                Iterator iter = channelBatches.entrySet().iterator();
+                while(iter.hasNext()) {
+                    Map.Entry<Long, EventBatch> pair = (Map.Entry) iter.next();
+                    EventBatch batch = pair.getValue();
+                    totalOutstandingEventBatches.decrementAndGet();
+                    batch.fail();
+                    expired.add(batch);
+                    iter.remove();
+                }
+                pollerCallback.onEventFailure(expired, new HecException("sticky_session_expired"));
+            }
+        }
+        outstandingEventBatches.remove(channel);
+        channel.setId();
+        String newChannelId = channel.getId();
+        log.info("Changed channel id from {} to {}", oldChannelId, newChannelId);
+
+        channel.setAvailable(true);
+        log.info("Channel {} is available", newChannelId);
+    }
+
     private void poll() {
         if (totalOutstandingEventBatches.get() <= 0 || outstandingEventBatches.size() <= 0) {
             return;
@@ -273,19 +319,22 @@ private void handleAckPollResult(HecChannel channel, HecAckPollResponse result)
 
         List<EventBatch> committedBatches = new ArrayList<>();
         ConcurrentHashMap<Long, EventBatch> channelBatches = outstandingEventBatches.get(channel);
-        for (Long id: ids) {
-            EventBatch batch = channelBatches.remove(id);
-            if (batch == null) {
-                log.warn("event batch id={} for channel={} on host={} is not in map anymore", id, channel, channel.getIndexer());
-                continue;
+        // Added null check as channelBatches might still be null(It may be removed while handling sticky sessions and not added until we send more data)
+        if (channelBatches != null) {
+            for (Long id: ids) {
+                EventBatch batch = channelBatches.remove(id);
+                if (batch == null) {
+                    log.warn("event batch id={} for channel={} on host={} is not in map anymore", id, channel, channel.getIndexer());
+                    continue;
+                }
+                totalOutstandingEventBatches.decrementAndGet();
+                batch.commit();
+                committedBatches.add(batch);
             }
-            totalOutstandingEventBatches.decrementAndGet();
-            batch.commit();
-            committedBatches.add(batch);
-        }
 
-        if (!committedBatches.isEmpty() && pollerCallback != null) {
-            pollerCallback.onEventCommitted(committedBatches);
+            if (!committedBatches.isEmpty() && pollerCallback != null) {
+                pollerCallback.onEventCommitted(committedBatches);
+            }
         }
     }
 
diff --git a/src/main/java/com/splunk/hecclient/HecChannel.java b/src/main/java/com/splunk/hecclient/HecChannel.java
@@ -24,10 +24,12 @@ final class HecChannel {
     private String id;
     private Map<String, String> chField;
     private IndexerInf indexer;
+    private boolean isAvailable;
 
     public HecChannel(IndexerInf idx) {
         id = newChannelId();
         indexer = idx;
+        isAvailable = true;
     }
 
     public IndexerInf getIndexer() {
@@ -48,6 +50,10 @@ public HecChannel setTracking(boolean trackChannel) {
         return this;
     }
 
+    public void setId() { id = newChannelId(); }
+
+    public void setAvailable(boolean isAvailable) { this.isAvailable = isAvailable; }
+
     public void send(final EventBatch batch) {
         if (chField != null) {
             batch.addExtraFields(chField);
@@ -60,9 +66,9 @@ public String executeHttpRequest(final HttpUriRequest req) {
         return indexer.executeHttpRequest(req);
     }
 
-    public boolean hasBackPressure() {
-        return indexer.hasBackPressure();
-    }
+    public boolean hasBackPressure() { return indexer.hasBackPressure(); }
+
+    public boolean isNotAvailable() { return isAvailable == false; }
 
     @Override
     public boolean equals(Object obj) {
diff --git a/src/main/java/com/splunk/hecclient/HttpClientBuilder.java b/src/main/java/com/splunk/hecclient/HttpClientBuilder.java
@@ -53,7 +53,7 @@ public HttpClientBuilder setSocketTimeout(int timeout /*seconds*/) {
     }
 
     public HttpClientBuilder setSocketSendBufferSize(int bufSize /*bytes*/) {
-        this.socketSendBufferSize = socketSendBufferSize;
+        this.socketSendBufferSize = bufSize;
         return this;
     }
 
diff --git a/src/main/java/com/splunk/hecclient/Indexer.java b/src/main/java/com/splunk/hecclient/Indexer.java
@@ -29,6 +29,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.util.concurrent.ConcurrentHashMap;
 
 final class Indexer implements IndexerInf {
     private static final Logger log = LoggerFactory.getLogger(Indexer.class);
@@ -124,9 +125,9 @@ public boolean send(final EventBatch batch) {
         }
 
         // we are all good
-        poller.add(channel, batch, resp);
-        log.debug("sent {} events to splunk through channel={} indexer={}",
-                batch.size(), channel.getId(), getBaseUrl());
+        poller.add(this.channel, batch, resp);
+        log.debug("sent {} events to splunk through channel={} indexer={}", batch.size(), channel.getId(), getBaseUrl());
+
         return true;
     }
 
@@ -161,7 +162,13 @@ private String readAndCloseResponse(CloseableHttpResponse resp) {
             }
         }
 
-        // log.info("event posting, channel={}, cookies={}", channel, resp.getHeaders("Set-Cookie"));
+//      log.info("event posting, channel={}, cookies={}, cookies.length={}", channel, resp.getHeaders("Set-Cookie"), resp.getHeaders("Set-Cookie").length);
+
+        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);
+        }
+
         int status = resp.getStatusLine().getStatusCode();
         // FIXME 503 server is busy backpressure
         if (status != 200 && status != 201) {
diff --git a/src/main/java/com/splunk/hecclient/LoadBalancer.java b/src/main/java/com/splunk/hecclient/LoadBalancer.java
@@ -52,7 +52,7 @@ public void send(final EventBatch batch) {
         for (int tried = 0; tried != channels.size(); tried++) {
             HecChannel channel = channels.get(index);
             index = (index + 1) % channels.size();
-            if (!channel.hasBackPressure()) {
+            if (!channel.hasBackPressure() && !channel.isNotAvailable()) {
                 channel.send(batch);
                 return;
             }
diff --git a/src/main/java/com/splunk/hecclient/Poller.java b/src/main/java/com/splunk/hecclient/Poller.java
@@ -20,7 +20,7 @@ public interface Poller {
     void stop();
     void add(HecChannel channel, EventBatch batch, String response);
     void fail(HecChannel channel, EventBatch batch, Exception ex);
-
+    void stickySessionHandler(HecChannel channel);
     // minimum load channel
     HecChannel getMinLoadChannel();
     long getTotalOutstandingEventBatches();
diff --git a/src/main/java/com/splunk/hecclient/ResponsePoller.java b/src/main/java/com/splunk/hecclient/ResponsePoller.java
@@ -31,6 +31,9 @@ public ResponsePoller(PollerCallback callback) {
         this.callback = callback;
     }
 
+    @Override
+    public void stickySessionHandler(HecChannel channel) {}
+
     @Override
     public void start() {
     }
diff --git a/src/main/java/com/splunk/kafka/connect/JacksonStructModule.java b/src/main/java/com/splunk/kafka/connect/JacksonStructModule.java
@@ -0,0 +1,44 @@
+/*
+ * Copyright 2017 Splunk, Inc..
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.splunk.kafka.connect;
+
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.JsonSerializer;
+import com.fasterxml.jackson.databind.SerializerProvider;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import org.apache.kafka.connect.data.Field;
+import org.apache.kafka.connect.data.Struct;
+
+import java.io.IOException;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+public class JacksonStructModule extends SimpleModule {
+  public JacksonStructModule() {
+    addSerializer(Struct.class, new StructSerializer());
+  }
+
+  static class StructSerializer extends JsonSerializer<Struct> {
+    @Override
+    public void serialize(Struct struct, JsonGenerator jsonGenerator, SerializerProvider serializerProvider) throws IOException {
+      final Map<String, Object> result = new LinkedHashMap<>(struct.schema().fields().size());
+      for (Field field : struct.schema().fields()) {
+        result.put(field.name(), struct.get(field));
+      }
+      jsonGenerator.writeObject(result);
+    }
+  }
+}
diff --git a/src/test/java/com/splunk/hecclient/HecAckPollerTest.java b/src/test/java/com/splunk/hecclient/HecAckPollerTest.java
@@ -323,4 +323,36 @@ public void getMinLoadChannel() {
 
         poller.stop();
     }
+
+    @Test
+    public void stickySessionHandler() {
+        PollerCallbackMock cb = new PollerCallbackMock();
+        HecAckPoller poller = new HecAckPoller(cb);
+        poller.setAckPollThreads(1);
+        poller.setAckPollInterval(2);
+        poller.start();
+
+        IndexerMock indexer = new IndexerMock();
+        String ackResponse = "{\"acks\":{\"1\":true}}";
+        indexer.setResponse(ackResponse);
+
+        HecChannel ch = new HecChannel(indexer);
+        EventBatch batch = UnitUtil.createBatch();
+
+        String response = "{\"text\":\"Success\",\"code\":0,\"ackId\":1}";
+        poller.add(ch, batch, response);
+
+        long outstanding = poller.getTotalOutstandingEventBatches();
+        Assert.assertEquals(1, outstanding);
+        UnitUtil.milliSleep(3000);
+
+        String oldId = ch.getId();
+        poller.stickySessionHandler(ch);
+        Assert.assertNotEquals(oldId, ch.getId());
+
+        outstanding = poller.getTotalOutstandingEventBatches();
+        Assert.assertEquals(0, outstanding);
+
+        poller.stop();
+    }
 }
diff --git a/src/test/java/com/splunk/hecclient/HecChannelTest.java b/src/test/java/com/splunk/hecclient/HecChannelTest.java
@@ -36,6 +36,18 @@ public void getterSetter() {
 
         Assert.assertEquals(id, ch.toString());
         Assert.assertNotNull(ch.hashCode());
+
+        Assert.assertFalse(ch.isNotAvailable());
+        ch.setAvailable(true);
+        Assert.assertFalse(ch.isNotAvailable());
+        ch.setAvailable(false);
+        Assert.assertTrue(ch.isNotAvailable());
+
+        ch.setId();
+        String newId = ch.getId();
+        Assert.assertNotNull(newId);
+        Assert.assertFalse(newId.isEmpty());
+        Assert.assertNotEquals(id, newId);
     }
 
     @Test
diff --git a/src/test/java/com/splunk/hecclient/LoadBalancerTest.java b/src/test/java/com/splunk/hecclient/LoadBalancerTest.java
@@ -102,6 +102,32 @@ public void sendWithOneBackPressure() {
         Assert.assertEquals(6, indexers.get(2).getBatches().size());
     }
 
+    @Test
+    public void sendWithOneNotAvailable() {
+        LoadBalancer lb = new LoadBalancer();
+        List<IndexerMock> indexers = new ArrayList<>();
+
+        int numberOfChannels = 3;
+        for (int i = 0; i < numberOfChannels; i++) {
+            IndexerMock indexer = new IndexerMock();
+            indexers.add(indexer);
+            HecChannel ch = new HecChannel(indexer);
+            if(i == 0) {
+                ch.setAvailable(false);
+            }
+            lb.add(ch);
+        }
+
+        int numberOfBatches = 12;
+        for (int i = 0; i < numberOfBatches; i++) {
+            lb.send(UnitUtil.createBatch());
+        }
+
+        Assert.assertEquals(0, indexers.get(0).getBatches().size());
+        Assert.assertEquals(6, indexers.get(1).getBatches().size());
+        Assert.assertEquals(6, indexers.get(2).getBatches().size());
+    }
+
     @Test(expected = HecException.class)
     public void sendWithoutChannels() {
         LoadBalancer lb = new LoadBalancer();
diff --git a/src/test/java/com/splunk/hecclient/PollerMock.java b/src/test/java/com/splunk/hecclient/PollerMock.java
diff --git a/src/test/java/com/splunk/kafka/connect/StructEventTest.java b/src/test/java/com/splunk/kafka/connect/StructEventTest.java

Original file line number	Diff line number	Diff line change
`@@ -53,7 +53,7 @@ public HttpClientBuilder setSocketTimeout(int timeout /seconds/) {`
`53`	`53`	`}`
`54`	`54`
`55`	`55`	`public HttpClientBuilder setSocketSendBufferSize(int bufSize /bytes/) {`
`56`		`- this.socketSendBufferSize = socketSendBufferSize;`
	`56`	`+ this.socketSendBufferSize = bufSize;`
`57`	`57`	`return this;`
`58`	`58`	`}`
`59`	`59`
Original file line number	Diff line number	Diff line change
`@@ -52,7 +52,7 @@ public void send(final EventBatch batch) {`
`52`	`52`	`for (int tried = 0; tried != channels.size(); tried++) {`
`53`	`53`	`HecChannel channel = channels.get(index);`
`54`	`54`	`index = (index + 1) % channels.size();`
`55`		`- if (!channel.hasBackPressure()) {`
	`55`	`+ if (!channel.hasBackPressure() && !channel.isNotAvailable()) {`
`56`	`56`	`channel.send(batch);`
`57`	`57`	`return;`
`58`	`58`	`}`