taskConfig) {
super(conf(), taskConfig);
splunkToken = getPassword(TOKEN_CONF).value();
@@ -201,55 +231,68 @@ public final class SplunkSinkConnectorConfig extends AbstractConfig {
lineBreaker = getString(LINE_BREAKER_CONF);
maxOutstandingEvents = getInt(MAX_OUTSTANDING_EVENTS_CONF);
maxRetries = getInt(MAX_RETRIES_CONF);
+ hecEventFormatted = getBoolean(HEC_EVENT_FORMATTED_CONF);
topicMetas = initMetaMap(taskConfig);
+ headerSupport = getBoolean(HEADER_SUPPORT_CONF);
+ headerCustom = getString(HEADER_CUSTOM_CONF);
+ headerIndex = getString(HEADER_INDEX_CONF);
+ headerSource = getString(HEADER_SOURCE_CONF);
+ headerSourcetype = getString(HEADER_SOURCETYPE_CONF);
+ headerHost = getString(HEADER_HOST_CONF);
}
public static ConfigDef conf() {
return new ConfigDef()
- .define(TOKEN_CONF, ConfigDef.Type.PASSWORD, ConfigDef.Importance.HIGH, TOKEN_DOC)
- .define(URI_CONF, ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, URI_DOC)
- .define(RAW_CONF, ConfigDef.Type.BOOLEAN, false, ConfigDef.Importance.MEDIUM, RAW_DOC)
- .define(ACK_CONF, ConfigDef.Type.BOOLEAN, false, ConfigDef.Importance.MEDIUM, ACK_DOC)
- .define(INDEX_CONF, ConfigDef.Type.STRING, "", ConfigDef.Importance.MEDIUM, INDEX_DOC)
- .define(SOURCETYPE_CONF, ConfigDef.Type.STRING, "", ConfigDef.Importance.MEDIUM, SOURCETYPE_DOC)
- .define(SOURCE_CONF, ConfigDef.Type.STRING, "", ConfigDef.Importance.MEDIUM, SOURCE_DOC)
- .define(HTTP_KEEPALIVE_CONF, ConfigDef.Type.BOOLEAN, true, ConfigDef.Importance.MEDIUM, HTTP_KEEPALIVE_DOC)
- .define(SSL_VALIDATE_CERTIFICATES_CONF, ConfigDef.Type.BOOLEAN, true, ConfigDef.Importance.MEDIUM, SSL_VALIDATE_CERTIFICATES_DOC)
- .define(SSL_TRUSTSTORE_PATH_CONF, ConfigDef.Type.STRING, "", ConfigDef.Importance.HIGH, SSL_TRUSTSTORE_PATH_DOC)
- .define(SSL_TRUSTSTORE_PASSWORD_CONF, ConfigDef.Type.PASSWORD, "", ConfigDef.Importance.HIGH, SSL_TRUSTSTORE_PASSWORD_DOC)
- .define(EVENT_TIMEOUT_CONF, ConfigDef.Type.INT, 300, ConfigDef.Importance.MEDIUM, EVENT_TIMEOUT_DOC)
- .define(ACK_POLL_INTERVAL_CONF, ConfigDef.Type.INT, 10, ConfigDef.Importance.MEDIUM, ACK_POLL_INTERVAL_DOC)
- .define(ACK_POLL_THREADS_CONF, ConfigDef.Type.INT, 2, ConfigDef.Importance.MEDIUM, ACK_POLL_THREADS_DOC)
- .define(MAX_HTTP_CONNECTION_PER_CHANNEL_CONF, ConfigDef.Type.INT, 2, ConfigDef.Importance.MEDIUM, MAX_HTTP_CONNECTION_PER_CHANNEL_DOC)
- .define(TOTAL_HEC_CHANNEL_CONF, ConfigDef.Type.INT, 2, ConfigDef.Importance.HIGH, TOTAL_HEC_CHANNEL_DOC)
- .define(SOCKET_TIMEOUT_CONF, ConfigDef.Type.INT, 60, ConfigDef.Importance.LOW, SOCKET_TIMEOUT_DOC)
- .define(ENRICHMENT_CONF, ConfigDef.Type.STRING, "", ConfigDef.Importance.LOW, ENRICHMENT_DOC)
- .define(TRACK_DATA_CONF, ConfigDef.Type.BOOLEAN, false, ConfigDef.Importance.LOW, TRACK_DATA_DOC)
- .define(USE_RECORD_TIMESTAMP_CONF, ConfigDef.Type.BOOLEAN, true, ConfigDef.Importance.MEDIUM, USE_RECORD_TIMESTAMP_DOC)
- .define(HEC_THREDS_CONF, ConfigDef.Type.INT, 1, ConfigDef.Importance.LOW, HEC_THREADS_DOC)
- .define(LINE_BREAKER_CONF, ConfigDef.Type.STRING, "", ConfigDef.Importance.MEDIUM, LINE_BREAKER_DOC)
- .define(MAX_OUTSTANDING_EVENTS_CONF, ConfigDef.Type.INT, 1000000, ConfigDef.Importance.MEDIUM, MAX_OUTSTANDING_EVENTS_DOC)
- .define(MAX_RETRIES_CONF, ConfigDef.Type.INT, -1, ConfigDef.Importance.MEDIUM, MAX_RETRIES_DOC)
- .define(MAX_BATCH_SIZE_CONF, ConfigDef.Type.INT, 500, ConfigDef.Importance.MEDIUM, MAX_BATCH_SIZE_DOC);
+ .define(TOKEN_CONF, ConfigDef.Type.PASSWORD, ConfigDef.Importance.HIGH, TOKEN_DOC)
+ .define(URI_CONF, ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, URI_DOC)
+ .define(RAW_CONF, ConfigDef.Type.BOOLEAN, false, ConfigDef.Importance.MEDIUM, RAW_DOC)
+ .define(ACK_CONF, ConfigDef.Type.BOOLEAN, false, ConfigDef.Importance.MEDIUM, ACK_DOC)
+ .define(INDEX_CONF, ConfigDef.Type.STRING, "", ConfigDef.Importance.MEDIUM, INDEX_DOC)
+ .define(SOURCETYPE_CONF, ConfigDef.Type.STRING, "", ConfigDef.Importance.MEDIUM, SOURCETYPE_DOC)
+ .define(SOURCE_CONF, ConfigDef.Type.STRING, "", ConfigDef.Importance.MEDIUM, SOURCE_DOC)
+ .define(HTTP_KEEPALIVE_CONF, ConfigDef.Type.BOOLEAN, true, ConfigDef.Importance.MEDIUM, HTTP_KEEPALIVE_DOC)
+ .define(SSL_VALIDATE_CERTIFICATES_CONF, ConfigDef.Type.BOOLEAN, true, ConfigDef.Importance.MEDIUM, SSL_VALIDATE_CERTIFICATES_DOC)
+ .define(SSL_TRUSTSTORE_PATH_CONF, ConfigDef.Type.STRING, "", ConfigDef.Importance.HIGH, SSL_TRUSTSTORE_PATH_DOC)
+ .define(SSL_TRUSTSTORE_PASSWORD_CONF, ConfigDef.Type.PASSWORD, "", ConfigDef.Importance.HIGH, SSL_TRUSTSTORE_PASSWORD_DOC)
+ .define(EVENT_TIMEOUT_CONF, ConfigDef.Type.INT, 300, ConfigDef.Importance.MEDIUM, EVENT_TIMEOUT_DOC)
+ .define(ACK_POLL_INTERVAL_CONF, ConfigDef.Type.INT, 10, ConfigDef.Importance.MEDIUM, ACK_POLL_INTERVAL_DOC)
+ .define(ACK_POLL_THREADS_CONF, ConfigDef.Type.INT, 2, ConfigDef.Importance.MEDIUM, ACK_POLL_THREADS_DOC)
+ .define(MAX_HTTP_CONNECTION_PER_CHANNEL_CONF, ConfigDef.Type.INT, 2, ConfigDef.Importance.MEDIUM, MAX_HTTP_CONNECTION_PER_CHANNEL_DOC)
+ .define(TOTAL_HEC_CHANNEL_CONF, ConfigDef.Type.INT, 2, ConfigDef.Importance.HIGH, TOTAL_HEC_CHANNEL_DOC)
+ .define(SOCKET_TIMEOUT_CONF, ConfigDef.Type.INT, 60, ConfigDef.Importance.LOW, SOCKET_TIMEOUT_DOC)
+ .define(ENRICHMENT_CONF, ConfigDef.Type.STRING, "", ConfigDef.Importance.LOW, ENRICHMENT_DOC)
+ .define(TRACK_DATA_CONF, ConfigDef.Type.BOOLEAN, false, ConfigDef.Importance.LOW, TRACK_DATA_DOC)
+ .define(USE_RECORD_TIMESTAMP_CONF, ConfigDef.Type.BOOLEAN, true, ConfigDef.Importance.MEDIUM, USE_RECORD_TIMESTAMP_DOC)
+ .define(HEC_THREDS_CONF, ConfigDef.Type.INT, 1, ConfigDef.Importance.LOW, HEC_THREADS_DOC)
+ .define(LINE_BREAKER_CONF, ConfigDef.Type.STRING, "", ConfigDef.Importance.MEDIUM, LINE_BREAKER_DOC)
+ .define(MAX_OUTSTANDING_EVENTS_CONF, ConfigDef.Type.INT, 1000000, ConfigDef.Importance.MEDIUM, MAX_OUTSTANDING_EVENTS_DOC)
+ .define(MAX_RETRIES_CONF, ConfigDef.Type.INT, -1, ConfigDef.Importance.MEDIUM, MAX_RETRIES_DOC)
+ .define(HEC_EVENT_FORMATTED_CONF, ConfigDef.Type.BOOLEAN, false, ConfigDef.Importance.LOW, HEC_EVENT_FORMATTED_DOC)
+ .define(MAX_BATCH_SIZE_CONF, ConfigDef.Type.INT, 500, ConfigDef.Importance.MEDIUM, MAX_BATCH_SIZE_DOC)
+ .define(HEADER_SUPPORT_CONF, ConfigDef.Type.BOOLEAN, false, ConfigDef.Importance.MEDIUM, HEADER_SUPPORT_DOC)
+ .define(HEADER_CUSTOM_CONF, ConfigDef.Type.STRING, "", ConfigDef.Importance.MEDIUM, HEADER_CUSTOM_DOC)
+ .define(HEADER_INDEX_CONF, ConfigDef.Type.STRING, "splunk.header.index", ConfigDef.Importance.MEDIUM, HEADER_INDEX_DOC)
+ .define(HEADER_SOURCE_CONF, ConfigDef.Type.STRING, "splunk.header.source", ConfigDef.Importance.MEDIUM, HEADER_SOURCE_DOC)
+ .define(HEADER_SOURCETYPE_CONF, ConfigDef.Type.STRING, "splunk.header.sourcetype", ConfigDef.Importance.MEDIUM, HEADER_SOURCETYPE_DOC)
+ .define(HEADER_HOST_CONF, ConfigDef.Type.STRING, "splunk.header.host", ConfigDef.Importance.MEDIUM, HEADER_HOST_DOC);
}
-
/**
- Configuration Method to setup all settings related to Splunk HEC Client
+ Configuration Method to setup all settings related to Splunk HEC Client
*/
public HecConfig getHecConfig() {
HecConfig config = new HecConfig(Arrays.asList(splunkURI.split(",")), splunkToken);
config.setDisableSSLCertVerification(!validateCertificates)
- .setSocketTimeout(socketTimeout)
- .setMaxHttpConnectionPerChannel(maxHttpConnPerChannel)
- .setTotalChannels(totalHecChannels)
- .setEventBatchTimeout(eventBatchTimeout)
- .setHttpKeepAlive(httpKeepAlive)
- .setAckPollInterval(ackPollInterval)
- .setAckPollThreads(ackPollThreads)
- .setEnableChannelTracking(trackData)
- .setTrustStorePath(trustStorePath)
- .setTrustStorePassword(trustStorePassword)
- .setHasCustomTrustStore(hasTrustStorePath);
+ .setSocketTimeout(socketTimeout)
+ .setMaxHttpConnectionPerChannel(maxHttpConnPerChannel)
+ .setTotalChannels(totalHecChannels)
+ .setEventBatchTimeout(eventBatchTimeout)
+ .setHttpKeepAlive(httpKeepAlive)
+ .setAckPollInterval(ackPollInterval)
+ .setAckPollThreads(ackPollThreads)
+ .setEnableChannelTracking(trackData)
+ .setTrustStorePath(trustStorePath)
+ .setTrustStorePassword(trustStorePassword)
+ .setHasCustomTrustStore(hasTrustStorePath);
return config;
}
@@ -266,6 +309,8 @@ public String toString() {
+ "indexes:" + indexes + ", "
+ "sourcetypes:" + sourcetypes + ", "
+ "sources:" + sources + ", "
+ + "headerSupport:" + headerSupport + ", "
+ + "headerCustom:" + headerCustom + ", "
+ "httpKeepAlive:" + httpKeepAlive + ", "
+ "validateCertificates:" + validateCertificates + ", "
+ "trustStorePath:" + trustStorePath + ", "
@@ -282,7 +327,14 @@ public String toString() {
+ "maxOutstandingEvents: " + maxOutstandingEvents + ", "
+ "maxRetries: " + maxRetries + ", "
+ "useRecordTimestamp: " + useRecordTimestamp + ", "
- + "trackData: " + trackData;
+ + "hecEventFormatted" + hecEventFormatted + ", "
+ + "trackData: " + trackData + ","
+ + "headerSupport:" + headerSupport + ","
+ + "headerCustom:" + headerCustom + ","
+ + "headerIndex:" + headerIndex + ","
+ + "headerSource:" + headerSource + ","
+ + "headerSourcetype:" + headerSourcetype + ","
+ + "headerHost" + headerHost;
}
private static String[] split(String data, String sep) {
diff --git a/src/main/java/com/splunk/kafka/connect/SplunkSinkRecord.java b/src/main/java/com/splunk/kafka/connect/SplunkSinkRecord.java
new file mode 100644
index 00000000..7a566115
--- /dev/null
+++ b/src/main/java/com/splunk/kafka/connect/SplunkSinkRecord.java
@@ -0,0 +1,172 @@
+/*
+ * Copyright 2017-2018 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 org.apache.commons.lang3.builder.HashCodeBuilder;
+import org.apache.commons.lang3.builder.EqualsBuilder;
+import org.apache.kafka.connect.header.Header;
+import org.apache.kafka.connect.header.Headers;
+import org.apache.kafka.connect.sink.SinkRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * SplunkSinkRecord provides helper functionality to enable Header support for the Splunk Connect for Kafka, Namely
+ * Header functionality introspection and comparison.
+ *
+ *
+ * @version 1.1.0
+ * @since 1.1.0
+ */
+public class SplunkSinkRecord {
+ private static final Logger log = LoggerFactory.getLogger(SplunkSinkRecord.class);
+ Headers headers;
+ SplunkSinkConnectorConfig connectorConfig;
+ String splunkHeaderIndex = "";
+ String splunkHeaderHost = "";
+ String splunkHeaderSource = "";
+ String splunkHeaderSourcetype = "";
+
+ public SplunkSinkRecord() {}
+
+ /**
+ * Creates a new Kafka Header utility object. Will take a Kafka SinkRecord and Splunk Sink Connector configuration
+ * and create the object based on Headers included with te Kafka Record.
+ *
+ * @param record Kafka SinkRecord to be introspected and headers retrieved from.
+ * @param connectorConfig Splunk Connector configuration used to determine headers of importance
+ * @version 1.1.0
+ * @since 1.1.0
+ */
+ public SplunkSinkRecord(SinkRecord record, SplunkSinkConnectorConfig connectorConfig) {
+ this.connectorConfig = connectorConfig;
+ this.headers = record.headers();
+ if(this.headers != null) {
+ setMetadataValues();
+ }
+ }
+
+ /**
+ * CompareRecordHeaders will compare a SinkRecords Header values against values that have already populate the
+ * Kakfa Header Utility object. This is used in batching events with the same meta-data values while using the /raw
+ * event point in Splunk
+ *
+ * @param record Kafka SinkRecord to be introspected and headers retrieved from.
+ * @version 1.1.0
+ * @since 1.1.0
+ */
+ protected boolean compareRecordHeaders(SinkRecord record) {
+ headers = record.headers();
+
+ Header indexHeader = headers.lastWithName(connectorConfig.headerIndex);
+ Header hostHeader = headers.lastWithName(connectorConfig.headerHost);
+ Header sourceHeader = headers.lastWithName(connectorConfig.headerSource);
+ Header sourcetypeHeader = headers.lastWithName(connectorConfig.headerSourcetype);
+
+ String index = "";
+ String host = "";
+ String source = "";
+ String sourcetype = "";
+
+ if(indexHeader != null) {
+ index = indexHeader.value().toString();
+ }
+ if(hostHeader != null) {
+ host = hostHeader.value().toString();
+ }
+ if(sourceHeader != null) {
+ source = sourceHeader.value().toString();
+ }
+ if(sourcetypeHeader != null) {
+ sourcetype = sourcetypeHeader.value().toString();
+ }
+
+ return splunkHeaderIndex.equals(index) && splunkHeaderHost.equals(host) &&
+ splunkHeaderSource.equals(source) && splunkHeaderSourcetype.equals(sourcetype);
+ }
+
+ private void setMetadataValues() {
+ Header indexHeader = this.headers.lastWithName(connectorConfig.headerIndex);
+ Header hostHeader = this.headers.lastWithName(connectorConfig.headerHost);
+ Header sourceHeader = this.headers.lastWithName(connectorConfig.headerSource);
+ Header sourcetypeHeader = this.headers.lastWithName(connectorConfig.headerSourcetype);
+
+ if(indexHeader != null) {
+ splunkHeaderIndex = indexHeader.value().toString();
+ }
+ if(hostHeader != null) {
+ splunkHeaderHost = hostHeader.value().toString();
+ }
+ if(sourceHeader != null) {
+ splunkHeaderSource = sourceHeader.value().toString();
+ }
+ if(sourcetypeHeader != null) {
+ splunkHeaderSourcetype = sourcetypeHeader.value().toString();
+ }
+ }
+
+ public String id() {
+ String separator = "$$$";
+ return new StringBuilder()
+ .append(splunkHeaderIndex)
+ .append(separator)
+ .append(splunkHeaderHost)
+ .append(separator)
+ .append(splunkHeaderSource)
+ .append(separator)
+ .append(splunkHeaderSourcetype)
+ .toString();
+ }
+
+ @Override
+ public int hashCode() {
+ return new HashCodeBuilder()
+ .append(splunkHeaderIndex)
+ .append(splunkHeaderHost)
+ .append(splunkHeaderSource)
+ .append(splunkHeaderSourcetype)
+ .toHashCode();
+ }
+
+ @Override
+ public boolean equals(Object obj) {
+ if (obj instanceof SplunkSinkRecord) {
+ final SplunkSinkRecord other = (SplunkSinkRecord) obj;
+ return id().equals(other.id());
+ }
+ return false;
+ }
+
+ public Headers getHeaders() {
+ return headers;
+ }
+
+ public String getSplunkHeaderIndex() {
+ return splunkHeaderIndex;
+ }
+
+ public String getSplunkHeaderHost() {
+ return splunkHeaderHost;
+ }
+
+ public String getSplunkHeaderSource() {
+ return splunkHeaderSource;
+ }
+
+ public String getSplunkHeaderSourcetype() {
+ return splunkHeaderSourcetype;
+ }
+}
\ No newline at end of file
diff --git a/src/main/java/com/splunk/kafka/connect/SplunkSinkTask.java b/src/main/java/com/splunk/kafka/connect/SplunkSinkTask.java
index 8a768686..9b2c4dae 100644
--- a/src/main/java/com/splunk/kafka/connect/SplunkSinkTask.java
+++ b/src/main/java/com/splunk/kafka/connect/SplunkSinkTask.java
@@ -16,12 +16,14 @@
package com.splunk.kafka.connect;
import com.splunk.hecclient.*;
-import com.splunk.kafka.connect.VersionUtils;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.connect.errors.RetriableException;
+import org.apache.kafka.connect.header.Headers;
import org.apache.kafka.connect.sink.SinkRecord;
import org.apache.kafka.connect.sink.SinkTask;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.kafka.connect.header.Header;
import java.util.*;
@@ -31,6 +33,7 @@
public final class SplunkSinkTask extends SinkTask implements PollerCallback {
private static final Logger log = LoggerFactory.getLogger(SplunkSinkTask.class);
private static final long flushWindow = 30 * 1000; // 30 seconds
+ private static final String HEADERTOKEN = "$$$";
private HecInf hec;
private KafkaRecordTracker tracker;
@@ -139,7 +142,9 @@ private void preventTooManyOutstandingEvents() {
}
private void handleRaw(final Collection records) {
- if (connectorConfig.hasMetaDataConfigured()) {
+ if(connectorConfig.headerSupport) {
+ if(records != null) { handleRecordsWithHeader(records); }
+ } else if (connectorConfig.hasMetaDataConfigured()) {
// when setup metadata - index, source, sourcetype, we need partition records for /raw
Map> partitionedRecords = partitionRecords(records);
for (Map.Entry> entry: partitionedRecords.entrySet()) {
@@ -152,6 +157,69 @@ private void handleRaw(final Collection records) {
}
}
+ private void handleRecordsWithHeader(final Collection records) {
+ HashMap> recordsWithSameHeaders = new HashMap<>();
+
+ for (SinkRecord record : records) {
+ String key = headerId(record);
+ if (!recordsWithSameHeaders.containsKey(key)) {
+ ArrayList recordList = new ArrayList();
+ recordsWithSameHeaders.put(key, recordList);
+ }
+ ArrayList recordList = recordsWithSameHeaders.get(key);
+ recordList.add(record);
+ }
+
+ Iterator>> itr = recordsWithSameHeaders.entrySet().iterator();
+ while(itr.hasNext()) {
+ Map.Entry set = itr.next();
+ String splunkSinkRecordKey = (String)set.getKey();
+ ArrayList recordArrayList = (ArrayList)set.getValue();
+ EventBatch batch = createRawHeaderEventBatch(splunkSinkRecordKey);
+ sendEvents(recordArrayList, batch);
+ }
+ log.debug("{} records have been bucketed in to {} batches", records.size(), recordsWithSameHeaders.size());
+ }
+
+ public String headerId(SinkRecord sinkRecord) {
+ Headers headers = sinkRecord.headers();
+
+ Header indexHeader = headers.lastWithName(connectorConfig.headerIndex);
+ Header hostHeader = headers.lastWithName(connectorConfig.headerHost);
+ Header sourceHeader = headers.lastWithName(connectorConfig.headerSource);
+ Header sourcetypeHeader = headers.lastWithName(connectorConfig.headerSourcetype);
+
+ StringBuilder headerString = new StringBuilder();
+
+ if(indexHeader != null) {
+ headerString.append(indexHeader.value().toString());
+ }
+
+ headerString.append(insertHeaderToken());
+
+ if(hostHeader != null) {
+ headerString.append(hostHeader.value().toString());
+ }
+
+ headerString.append(insertHeaderToken());
+
+ if(sourceHeader != null) {
+ headerString.append(sourceHeader.value().toString());
+ }
+
+ headerString.append(insertHeaderToken());
+
+ if(sourcetypeHeader != null) {
+ headerString.append(sourcetypeHeader.value().toString());
+ }
+
+ return headerString.toString();
+ }
+
+ public String insertHeaderToken() {
+ return HEADERTOKEN;
+ }
+
private void handleEvent(final Collection records) {
EventBatch batch = new JsonEventBatch();
sendEvents(records, batch);
@@ -194,6 +262,17 @@ private void send(final EventBatch batch) {
}
}
+ private EventBatch createRawHeaderEventBatch(String splunkSinkRecord) {
+ String[] split = splunkSinkRecord.split("[$]{3}");
+
+ return RawEventBatch.factory()
+ .setIndex(split[0])
+ .setSourcetype(split[1])
+ .setSource(split[2])
+ .setHost(split[3])
+ .build();
+ }
+
// setup metadata on RawEventBatch
private EventBatch createRawEventBatch(final TopicPartition tp) {
if (tp == null) {
@@ -235,7 +314,7 @@ public String version() {
public void onEventCommitted(final List batches) {
// for (final EventBatch batch: batches) {
- // assert batch.isCommitted();
+ // assert batch.isCommitted();
// }
}
@@ -250,26 +329,31 @@ private Event createHecEventFrom(final SinkRecord record) {
if (connectorConfig.raw) {
RawEvent event = new RawEvent(record.value(), record);
event.setLineBreaker(connectorConfig.lineBreaker);
+ if(connectorConfig.headerSupport) {
+ event = (RawEvent)addHeaders(event, record);
+ }
return event;
}
- // meta data for /event endpoint is per event basis
- JsonEvent event = new JsonEvent(record.value(), record);
- if (connectorConfig.useRecordTimestamp && record.timestamp() != null) {
- // record timestamp is in milliseconds
- event.setTime(record.timestamp() / 1000.0);
+ JsonEvent event = null;
+ ObjectMapper objectMapper = new ObjectMapper();
+
+ if(connectorConfig.hecEventFormatted) {
+ try {
+ event = objectMapper.readValue(record.value().toString(), JsonEvent.class);
+ } catch(Exception e) {
+ log.error("event does not follow correct HEC pre-formatted format", record.toString());
+ event = createHECEventNonFormatted(record);
+ }
+ } else {
+ event = createHECEventNonFormatted(record);
}
- Map metas = connectorConfig.topicMetas.get(record.topic());
- if (metas != null) {
- event.setIndex(metas.get(SplunkSinkConnectorConfig.INDEX));
- event.setSourcetype(metas.get(SplunkSinkConnectorConfig.SOURCETYPE));
- event.setSource(metas.get(SplunkSinkConnectorConfig.SOURCE));
- event.addFields(connectorConfig.enrichments);
+ if(connectorConfig.headerSupport) {
+ addHeaders(event, record);
}
if (connectorConfig.trackData) {
- // for data loss, latency tracking
Map trackMetas = new HashMap<>();
trackMetas.put("kafka_offset", String.valueOf(record.kafkaOffset()));
trackMetas.put("kafka_timestamp", String.valueOf(record.timestamp()));
@@ -277,12 +361,67 @@ private Event createHecEventFrom(final SinkRecord record) {
trackMetas.put("kafka_partition", String.valueOf(record.kafkaPartition()));
event.addFields(trackMetas);
}
-
event.validate();
return event;
}
+ private Event addHeaders(Event event, SinkRecord record) {
+ Headers headers = record.headers();
+ if(headers.isEmpty() && connectorConfig.headerCustom.isEmpty()) {
+ return event;
+ }
+
+ Header headerIndex = headers.lastWithName(connectorConfig.headerIndex);
+ Header headerHost = headers.lastWithName(connectorConfig.headerHost);
+ Header headerSource = headers.lastWithName(connectorConfig.headerSource);
+ Header headerSourcetype = headers.lastWithName(connectorConfig.headerSourcetype);
+
+ if (headerIndex != null) {
+ event.setIndex(headerIndex.value().toString());
+ }
+ if (headerHost != null) {
+ event.setHost(headerHost.value().toString());
+ }
+ if (headerSource != null) {
+ event.setSource(headerSource.value().toString());
+ }
+ if (headerSourcetype != null) {
+ event.setSourcetype(headerSourcetype.value().toString());
+ }
+
+ // Custom headers are configured with a comma separated list passed in configuration
+ // "custom_header_1,custom_header_2,custom_header_3"
+ if (!connectorConfig.headerCustom.isEmpty()) {
+ String[] customHeaders = connectorConfig.headerCustom.split(",");
+ Map headerMap = new HashMap<>();
+ for (String header : customHeaders) {
+ Header customHeader = headers.lastWithName(header);
+ if (customHeader != null) {
+ headerMap.put(header, customHeader.value().toString());
+ }
+ }
+ event.addFields(headerMap);
+ }
+ return event;
+ }
+
+ private JsonEvent createHECEventNonFormatted(final SinkRecord record) {
+ JsonEvent event = new JsonEvent(record.value(), record);
+ if (connectorConfig.useRecordTimestamp && record.timestamp() != null) {
+ event.setTime(record.timestamp() / 1000.0); // record timestamp is in milliseconds
+ }
+
+ Map metas = connectorConfig.topicMetas.get(record.topic());
+ if (metas != null) {
+ event.setIndex(metas.get(SplunkSinkConnectorConfig.INDEX));
+ event.setSourcetype(metas.get(SplunkSinkConnectorConfig.SOURCETYPE));
+ event.setSource(metas.get(SplunkSinkConnectorConfig.SOURCE));
+ event.addFields(connectorConfig.enrichments);
+ }
+ return event;
+ }
+
private Event createHecEventFromMalformed(final SinkRecord record) {
Object data;
if (connectorConfig.raw) {
diff --git a/src/main/resources/version.properties b/src/main/resources/version.properties
index f306651a..4ddaa874 100644
--- a/src/main/resources/version.properties
+++ b/src/main/resources/version.properties
@@ -1,3 +1,3 @@
-githash=@0f3e74e
-gitbranch=issue117-rename-directories
-gitversion=v1.0.0-LAR
+githash=@1e92a95
+gitbranch=release/1.0.x
+gitversion=v1.0.0
diff --git a/src/test/java/com/splunk/hecclient/RawEventBatchTest.java b/src/test/java/com/splunk/hecclient/RawEventBatchTest.java
index 57a2473d..495fd8d3 100644
--- a/src/test/java/com/splunk/hecclient/RawEventBatchTest.java
+++ b/src/test/java/com/splunk/hecclient/RawEventBatchTest.java
@@ -103,4 +103,23 @@ public void getter() {
Assert.assertEquals("index", batch.getIndex());
Assert.assertEquals(1, batch.getTime());
}
+
+ @Test
+ public void checkEquals() {
+ RawEventBatch batchOne = RawEventBatch.factory()
+ .setSource("source3")
+ .setIndex("idx1")
+ .setSourcetype("sourcetype2")
+ .setHost("host4")
+ .build();
+
+ RawEventBatch batchTwo = RawEventBatch.factory()
+ .setSource("source")
+ .setIndex("idx")
+ .setSourcetype("1sourcetype2")
+ .setHost("3host4")
+ .build();
+
+ Assert.assertFalse(batchOne.equals(batchTwo));
+ }
}
diff --git a/src/test/java/com/splunk/kafka/connect/ConfigProfile.java b/src/test/java/com/splunk/kafka/connect/ConfigProfile.java
index f49ea8ca..215e393a 100644
--- a/src/test/java/com/splunk/kafka/connect/ConfigProfile.java
+++ b/src/test/java/com/splunk/kafka/connect/ConfigProfile.java
@@ -28,6 +28,13 @@ public class ConfigProfile {
private boolean trackData;
private int maxBatchSize;
private int numOfThreads;
+ private boolean headerSupport;
+ private boolean hecFormatted;
+ private String headerCustom;
+ private String headerIndex;
+ private String headerSource;
+ private String headerSourcetype;
+ private String headerHost;
public ConfigProfile() {
this(0);
@@ -41,6 +48,8 @@ public ConfigProfile(int profile) {
break;
case 2: buildProfileTwo();
break;
+ case 3: buildProfileThree();
+ break;
default: buildProfileDefault();
break;
}
@@ -143,6 +152,37 @@ public ConfigProfile buildProfileTwo() {
return this;
}
+ public ConfigProfile buildProfileThree() {
+ this.topics = "kafka-data";
+ this.token = "mytoken";
+ this.uri = "https://dummy:8088";
+ this.raw = true;
+ this.ack = false;
+ this.indexes = "index-1";
+ this.sourcetypes = "kafka-data";
+ this.sources = "kafka-connect";
+ this.httpKeepAlive = true;
+ this.validateCertificates = false;
+ this.eventBatchTimeout = 1;
+ this.ackPollInterval = 1;
+ this.ackPollThreads = 1;
+ this.maxHttpConnPerChannel = 1;
+ this.totalHecChannels = 1;
+ this.socketTimeout = 1;
+ this.enrichements = "hello=world";
+ this.enrichementMap = new HashMap<>();
+ this.trackData = false;
+ this.maxBatchSize = 1;
+ this.numOfThreads = 1;
+ this.headerSupport = true;
+ this.headerIndex = "splunk.header.index";
+ this.headerSource = "splunk.header.source";
+ this.headerSourcetype = "splunk.header.sourcetype";
+ this.headerHost = "splunk.header.host";
+
+ return this;
+ }
+
public String getTopics() {
return topics;
}
@@ -335,6 +375,38 @@ public void setNumOfThreads(int numOfThreads) {
this.numOfThreads = numOfThreads;
}
+ public String getHeaderIndex() {
+ return headerIndex;
+ }
+
+ public void setHeaderIndex(String headerIndex) {
+ this.headerIndex = headerIndex;
+ }
+
+ public String getHeaderSource() {
+ return headerSource;
+ }
+
+ public void setHeaderSource(String headerSource) {
+ this.headerSource = headerSource;
+ }
+
+ public String getHeaderSourcetype() {
+ return headerSourcetype;
+ }
+
+ public void setHeaderSourcetype(String headerSourcetype) {
+ this.headerSourcetype = headerSourcetype;
+ }
+
+ public String getHeaderHost() {
+ return headerHost;
+ }
+
+ public void setHeaderHost(String headerHost) {
+ this.headerHost = headerHost;
+ }
+
@Override public String toString() {
return "ConfigProfile{" + "topics='" + topics + '\'' + ", token='" + token + '\'' + ", uri='" + uri + '\'' + ", raw=" + raw + ", ack=" + ack + ", indexes='" + indexes + '\'' + ", sourcetypes='" + sourcetypes + '\'' + ", sources='" + sources + '\'' + ", httpKeepAlive=" + httpKeepAlive + ", validateCertificates=" + validateCertificates + ", hasTrustStorePath=" + hasTrustStorePath + ", trustStorePath='" + trustStorePath + '\'' + ", trustStorePassword='" + trustStorePassword + '\'' + ", eventBatchTimeout=" + eventBatchTimeout + ", ackPollInterval=" + ackPollInterval + ", ackPollThreads=" + ackPollThreads + ", maxHttpConnPerChannel=" + maxHttpConnPerChannel + ", totalHecChannels=" + totalHecChannels + ", socketTimeout=" + socketTimeout + ", enrichements='" + enrichements + '\'' + ", enrichementMap=" + enrichementMap + ", trackData=" + trackData + ", maxBatchSize=" + maxBatchSize + ", numOfThreads=" + numOfThreads + '}';
}
diff --git a/src/test/java/com/splunk/kafka/connect/SplunkSinkRecordTest.java b/src/test/java/com/splunk/kafka/connect/SplunkSinkRecordTest.java
new file mode 100644
index 00000000..a8a58c65
--- /dev/null
+++ b/src/test/java/com/splunk/kafka/connect/SplunkSinkRecordTest.java
@@ -0,0 +1,107 @@
+/*
+ * Copyright 2017-2018 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 org.apache.kafka.connect.header.Headers;
+import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.sink.SinkRecord;
+import org.junit.Assert;
+import org.junit.jupiter.api.Test;
+
+import java.util.Map;
+
+public class SplunkSinkRecordTest {
+
+ @Test
+ public void checkKafkaHeaderUtilityGetters() {
+ UnitUtil uu = new UnitUtil(3);
+ Map config = uu.createTaskConfig();
+
+ SplunkSinkConnectorConfig connectorConfig = new SplunkSinkConnectorConfig(config);
+
+ SinkRecord record = setupRecord();
+ Headers headers = record.headers();
+
+ headers.addString(uu.configProfile.getHeaderIndex(), "splunk.header.index");
+ headers.addString(uu.configProfile.getHeaderHost(), "splunk.header.host");
+ headers.addString(uu.configProfile.getHeaderSource(), "splunk.header.source");
+ headers.addString(uu.configProfile.getHeaderSourcetype(), "splunk.header.sourcetype");
+
+ System.out.println(headers.toString());
+
+ SplunkSinkRecord splunkSinkRecord = new SplunkSinkRecord(record, connectorConfig);
+
+ Assert.assertEquals("splunk.header.index", (splunkSinkRecord.getSplunkHeaderIndex()));
+ Assert.assertEquals("splunk.header.host", (splunkSinkRecord.getSplunkHeaderHost()));
+ Assert.assertEquals("splunk.header.source", (splunkSinkRecord.getSplunkHeaderSource()));
+ Assert.assertEquals("splunk.header.sourcetype", (splunkSinkRecord.getSplunkHeaderSourcetype()));
+ }
+
+ @Test
+ public void CompareRecordHeaders() {
+ UnitUtil uu = new UnitUtil(3);
+ Map config = uu.createTaskConfig();
+
+ SinkRecord record_1 = setupRecord();
+
+ Headers headers_1 = record_1.headers();
+ headers_1.addString("splunk.header.index", "header-index");
+ headers_1.addString("splunk.header.host", "header.splunk.com");
+ headers_1.addString("splunk.header.source", "headersource");
+ headers_1.addString("splunk.header.sourcetype", "test message");
+
+ SplunkSinkConnectorConfig connectorConfig = new SplunkSinkConnectorConfig(config);
+
+ SplunkSinkRecord splunkSinkRecord = new SplunkSinkRecord(record_1, connectorConfig);
+
+ SinkRecord record_2 = setupRecord();
+
+ Headers headers_2 = record_2.headers();
+ headers_2.addString("splunk.header.index", "header-index");
+ headers_2.addString("splunk.header.host", "header.splunk.com");
+ headers_2.addString("splunk.header.source", "headersource");
+ headers_2.addString("splunk.header.sourcetype", "test message");
+
+ Assert.assertTrue(splunkSinkRecord.compareRecordHeaders(record_2));
+
+ SinkRecord record_3 = setupRecord();
+
+ Headers headers_3 = record_3.headers();
+ headers_3.addString("splunk.header.index", "header-index=diff");
+ headers_3.addString("splunk.header.host", "header.splunk.com");
+ headers_3.addString("splunk.header.source", "headersource");
+ headers_3.addString("splunk.header.sourcetype", "test message");
+
+ Assert.assertFalse(splunkSinkRecord.compareRecordHeaders(record_3));
+ }
+
+ public SinkRecord setupRecord() {
+ String topic = "test-topic";
+ int partition = 1;
+ Schema keySchema = null;
+ Object key = "key";
+ Schema valueSchema = null;
+ Object value = "value";
+ long timestamp = System.currentTimeMillis();
+
+ SinkRecord record = createMockSinkRecord(topic, partition, keySchema, key, valueSchema, value, timestamp);
+ return record;
+ }
+
+ public SinkRecord createMockSinkRecord(String topic, int partition, Schema keySchema, Object key, Schema valueSchema, Object value, long timestamp) {
+ return new SinkRecord(topic, partition, keySchema, key, valueSchema, value, timestamp);
+ }
+}
\ No newline at end of file