Ignore failed batch with 'invalid event format' error (#242)

Author: rockb1017

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

@@ -130,13 +130,22 @@ public void add(HecChannel channel, EventBatch batch, String response) {
             fail(channel, batch, ex);
             return;
         }
+        
+        if (resp.getText() == "Invalid data format") {
+            log.warn("Invalid Splunk HEC data format. Ignoring events. channel={} index={} events={}", channel, channel.getIndexer(), batch.toString());
+            batch.commit();
+            List<EventBatch> committedBatches = new ArrayList<>();
+            committedBatches.add(batch);
+            pollerCallback.onEventCommitted(committedBatches);
+            return;
+        }
 
         ConcurrentHashMap<Long, EventBatch> channelEvents = outstandingEventBatches.get(channel);
         if (channelEvents == null) {
             outstandingEventBatches.putIfAbsent(channel, new ConcurrentHashMap<>());
             channelEvents = outstandingEventBatches.get(channel);
         }
-
+        
         if (channelEvents.get(resp.getAckId()) != null) {
             log.warn("ackId={} already exists for channel={} index={} data may be duplicated in Splunk", resp.getAckId(), channel, channel.getIndexer());
             return;

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

@@ -26,13 +26,18 @@
 import org.apache.http.message.BasicHeader;
 import org.apache.http.protocol.HttpContext;
 import org.apache.http.util.EntityUtils;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.fasterxml.jackson.databind.JsonNode;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 
 final class Indexer implements IndexerInf {
     private static final Logger log = LoggerFactory.getLogger(Indexer.class);
+    private static final ObjectMapper jsonMapper = new ObjectMapper();
 
     private CloseableHttpClient httpClient;
     private HttpContext context;
@@ -165,8 +170,8 @@ private String readAndCloseResponse(CloseableHttpResponse resp) {
                 throw new HecException("failed to close http response", ex);
             }
         }
-
-//      log.info("event posting, channel={}, cookies={}, cookies.length={}", channel, resp.getHeaders("Set-Cookie"), resp.getHeaders("Set-Cookie").length);
+        
+        //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");
@@ -181,7 +186,25 @@ private String readAndCloseResponse(CloseableHttpResponse resp) {
             }
 
             log.error("failed to post events resp={}, status={}", respPayload, status);
-            throw new HecException(String.format("failed to post events resp=%s, status=%d", respPayload, status));
+            JsonNode jsonNode;
+            try {
+                jsonNode = jsonMapper.readTree(respPayload);
+            } catch (Exception ex) {
+                log.error("failed to parse response payload", ex);
+                throw new HecException("failed to parse response payload", ex);
+            }
+            
+            String respText = (jsonNode.has("text")) ? jsonNode.get("text").asText() : null;
+
+            if (respText == "Invalid data format") {
+                ObjectNode objNode = jsonMapper.createObjectNode();
+                objNode.put("text", "Invalid data format");
+                objNode.put("code", 0); // Mark it as success
+                objNode.put("ackId", -1);
+                respPayload = objNode.toString();
+            } else {
+                throw new HecException(String.format("failed to post events resp=%s, status=%d", respPayload, status));
+            }
         }
 
         clearBackPressure();

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

@@ -68,6 +68,9 @@ public void add(HecChannel channel, EventBatch batch, String resp) {
                 fail(channel, batch, new HecException(response.getText()));
                 return;
             }
+            if (response.getText() == "Invalid data format") {
+                log.warn("Invalid Splunk HEC data format. Ignoring events. channel={} index={} events={}", channel, channel.getIndexer(), batch.toString());
+            }
         } catch (Exception ex) {
             log.error("failed to parse response", resp, ex);
             fail(channel, batch, ex);

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

@@ -30,6 +30,7 @@ public class CloseableHttpClientMock extends CloseableHttpClient {
     public static final String success = "{\"text\":\"Success\",\"code\":0,\"ackId\":2}";
     public static final String serverBusy = "{\"text\":\"Server busy\",\"code\":1}";
     public static final String noDataError = "{\"text\":\"No data\",\"code\":5}";
+    public static final String invalidDataFormat = "{\"text\":\"Invalid data format\",\"code\":6}";
     public static final String exception = "excpetion";
 
     private String resp = "";

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

@@ -112,6 +112,24 @@ public void sendWithSuccess() {
         }
     }
 
+    @Test
+    public void sendWithInvalidData() {
+        CloseableHttpClientMock client = new CloseableHttpClientMock();
+        client.setResponse(CloseableHttpClientMock.invalidDataFormat);
+        PollerMock poller = new PollerMock();
+
+        Indexer indexer = new Indexer(baseUrl, token, client, poller);
+        EventBatch batch = UnitUtil.createBatch();
+        boolean result = indexer.send(batch);
+        Assert.assertTrue(result);
+        Assert.assertNotNull(poller.getBatch());
+        Assert.assertNull(poller.getFailedBatch());
+        Assert.assertNull(poller.getException());
+        Assert.assertEquals(indexer.getChannel(), poller.getChannel());
+        Assert.assertEquals(CloseableHttpClientMock.success, poller.getResponse());
+    }
+
+
     @Test
     public void sendWithServerBusy() {
         CloseableHttpClientMock client = new CloseableHttpClientMock();