You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flume.apache.org by rg...@apache.org on 2022/04/04 08:44:24 UTC
[flume-search] branch master created (now 3d096e5)
This is an automated email from the ASF dual-hosted git repository.
rgoers pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flume-search.git
at 3d096e5 Fork the sink from main flume build
This branch includes the following new commits:
new 3d096e5 Fork the sink from main flume build
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[flume-search] 01/01: Fork the sink from main flume build
Posted by rg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
rgoers pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flume-search.git
commit 3d096e5634328892625d526f73eadf37af575684
Author: Ralph Goers <rg...@apache.org>
AuthorDate: Mon Apr 4 01:44:16 2022 -0700
Fork the sink from main flume build
---
.github/workflows/build.yml | 63 +++
.github/workflows/codeql-analysis.yml | 70 +++
.gitignore | 22 +
flume-elasticsearch-sink/pom.xml | 96 ++++
...actElasticSearchIndexRequestBuilderFactory.java | 124 +++++
.../sink/elasticsearch/ContentBuilderUtil.java | 86 ++++
.../ElasticSearchDynamicSerializer.java | 73 +++
.../ElasticSearchEventSerializer.java | 48 ++
.../ElasticSearchIndexRequestBuilderFactory.java | 60 +++
.../ElasticSearchLogStashEventSerializer.java | 145 ++++++
.../sink/elasticsearch/ElasticSearchSink.java | 434 ++++++++++++++++++
.../elasticsearch/ElasticSearchSinkConstants.java | 111 +++++
.../EventSerializerIndexRequestBuilderFactory.java | 69 +++
.../flume/sink/elasticsearch/IndexNameBuilder.java | 42 ++
.../sink/elasticsearch/SimpleIndexNameBuilder.java | 46 ++
.../elasticsearch/TimeBasedIndexNameBuilder.java | 92 ++++
.../flume/sink/elasticsearch/TimestampedEvent.java | 60 +++
.../elasticsearch/client/ElasticSearchClient.java | 58 +++
.../client/ElasticSearchClientFactory.java | 78 ++++
.../client/ElasticSearchRestClient.java | 149 ++++++
.../client/ElasticSearchTransportClient.java | 228 ++++++++++
.../client/NoSuchClientTypeException.java | 23 +
.../sink/elasticsearch/client/RoundRobinList.java | 44 ++
.../AbstractElasticSearchSinkTest.java | 164 +++++++
.../TestElasticSearchDynamicSerializer.java | 64 +++
...estElasticSearchIndexRequestBuilderFactory.java | 215 +++++++++
.../TestElasticSearchLogStashEventSerializer.java | 127 ++++++
.../sink/elasticsearch/TestElasticSearchSink.java | 505 +++++++++++++++++++++
.../TestElasticSearchSinkCreation.java | 49 ++
.../TimeBasedIndexNameBuilderTest.java | 60 +++
.../sink/elasticsearch/TimestampedEventTest.java | 88 ++++
.../elasticsearch/client/RoundRobinListTest.java | 42 ++
.../client/TestElasticSearchClientFactory.java | 64 +++
.../client/TestElasticSearchRestClient.java | 180 ++++++++
.../client/TestElasticSearchTransportClient.java | 127 ++++++
.../src/test/resources/log4j2.xml | 31 ++
pom.xml | 277 +++++++++++
37 files changed, 4214 insertions(+)
diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
new file mode 100644
index 0000000..96130aa
--- /dev/null
+++ b/.github/workflows/build.yml
@@ -0,0 +1,63 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You 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.
+
+name: build
+
+on:
+ push:
+ branches:
+ - trunk
+ pull_request:
+
+jobs:
+ build:
+
+ runs-on: ${{ matrix.os }}
+
+ strategy:
+ matrix:
+ os: [ ubuntu-latest, macos-latest ]
+
+ steps:
+
+ - name: Checkout repository
+ uses: actions/checkout@v2
+
+ # JDK 8 is needed for the build, and it is the primary bytecode target.
+ - name: Setup JDK 8
+ uses: actions/setup-java@v2.3.0
+ with:
+ distribution: temurin
+ java-version: 8
+ java-package: jdk
+ architecture: x64
+ cache: maven
+
+ - name: Inspect environment (Linux)
+ if: runner.os == 'Linux'
+ run: env | grep '^JAVA'
+
+ - name: Inspect environment (MacOS)
+ if: runner.os == 'macOS'
+ run: env | grep '^JAVA'
+
+ - name: Build with Maven
+ timeout-minutes: 120
+ shell: bash
+ run: |
+ ./mvnw clean verify -DredirectTestOutput=true \
+ --show-version --batch-mode --errors --no-transfer-progress \
+ -DtrimStackTrace=false \
+ -Dsurefire.rerunFailingTestsCount=2
diff --git a/.github/workflows/codeql-analysis.yml b/.github/workflows/codeql-analysis.yml
new file mode 100644
index 0000000..20d8be9
--- /dev/null
+++ b/.github/workflows/codeql-analysis.yml
@@ -0,0 +1,70 @@
+# For most projects, this workflow file will not need changing; you simply need
+# to commit it to your repository.
+#
+# You may wish to alter this file to override the set of languages analyzed,
+# or to provide custom queries or build logic.
+#
+# ******** NOTE ********
+# We have attempted to detect the languages in your repository. Please check
+# the `language` matrix defined below to confirm you have the correct set of
+# supported CodeQL languages.
+#
+name: "CodeQL"
+
+on:
+ push:
+ branches: [ trunk ]
+ pull_request:
+ # The branches below must be a subset of the branches above
+ branches: [ trunk ]
+ schedule:
+ - cron: '15 0 * * 5'
+
+jobs:
+ analyze:
+ name: Analyze
+ runs-on: ubuntu-latest
+ permissions:
+ actions: read
+ contents: read
+ security-events: write
+
+ strategy:
+ fail-fast: false
+ matrix:
+ language: [ 'java', 'python' ]
+ # CodeQL supports [ 'cpp', 'csharp', 'go', 'java', 'javascript', 'python', 'ruby' ]
+ # Learn more about CodeQL language support at https://git.io/codeql-language-support
+
+ steps:
+ - name: Checkout repository
+ uses: actions/checkout@v2
+
+ # Initializes the CodeQL tools for scanning.
+ - name: Initialize CodeQL
+ uses: github/codeql-action/init@v1
+ with:
+ languages: ${{ matrix.language }}
+ # If you wish to specify custom queries, you can do so here or in a config file.
+ # By default, queries listed here will override any specified in a config file.
+ # Prefix the list here with "+" to use these queries and those in the config file.
+ # queries: ./path/to/local/query, your-org/your-repo/queries@main
+
+ # Autobuild attempts to build any compiled languages (C/C++, C#, or Java).
+ # If this step fails, then you should remove it and run the build manually (see below)
+ - name: Autobuild
+ uses: github/codeql-action/autobuild@v1
+
+ # âšī¸ Command-line programs to run using the OS shell.
+ # đ https://git.io/JvXDl
+
+ # âī¸ If the Autobuild fails above, remove it and uncomment the following three lines
+ # and modify them (or add more) to build your code if your project
+ # uses a compiled language
+
+ #- run: |
+ # make bootstrap
+ # make release
+
+ - name: Perform CodeQL Analysis
+ uses: github/codeql-action/analyze@v1
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..9a0a932
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1,22 @@
+# Lines that start with '#' are comments.
+*~
+*.diff
+*#
+.classpath
+.project
+.settings
+bin/flume-env.sh
+conf/flume-site.xml
+bin/.settings
+.eclipse
+pmd_report.html
+*/bin
+target
+patchprocess
+derby.log
+.idea
+*.iml
+nb-configuration.xml
+.DS_Store
+/.mvn/wrapper/maven-wrapper.jar
+**/metastore_db
diff --git a/flume-elasticsearch-sink/pom.xml b/flume-elasticsearch-sink/pom.xml
new file mode 100644
index 0000000..a5beb8e
--- /dev/null
+++ b/flume-elasticsearch-sink/pom.xml
@@ -0,0 +1,96 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!-- Licensed to the Apache Software Foundation (ASF) under one or more contributor
+ license agreements. See the NOTICE file distributed with this work for additional
+ information regarding copyright ownership. The ASF licenses this file to
+ You 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. -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <artifactId>flume-search</artifactId>
+ <groupId>org.apache.flume</groupId>
+ <version>1.0.0-SNAPSHOT</version>
+ </parent>
+
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-elasticsearch-sink</artifactId>
+ <name>Apache Flume ElasticSearch Sink</name>
+
+ <properties>
+ <!-- TODO fix spotbugs/pmd violations -->
+ <spotbugs.maxAllowedViolations>8</spotbugs.maxAllowedViolations>
+ <pmd.maxAllowedViolations>10</pmd.maxAllowedViolations>
+ </properties>
+
+ <dependencies>
+
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-ng-sdk</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-ng-core</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.slf4j</groupId>
+ <artifactId>slf4j-api</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.elasticsearch</groupId>
+ <artifactId>elasticsearch</artifactId>
+ <optional>true</optional>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.httpcomponents</groupId>
+ <artifactId>httpclient</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>junit</groupId>
+ <artifactId>junit</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-slf4j-impl</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.logging.log4j</groupId>
+ <artifactId>log4j-1.2-api</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>commons-lang</groupId>
+ <artifactId>commons-lang</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>com.google.guava</groupId>
+ <artifactId>guava</artifactId>
+ </dependency>
+
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-all</artifactId>
+ <scope>test</scope>
+ </dependency>
+
+ </dependencies>
+</project>
diff --git a/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/AbstractElasticSearchIndexRequestBuilderFactory.java b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/AbstractElasticSearchIndexRequestBuilderFactory.java
new file mode 100644
index 0000000..754155c
--- /dev/null
+++ b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/AbstractElasticSearchIndexRequestBuilderFactory.java
@@ -0,0 +1,124 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch;
+
+import java.io.IOException;
+
+import org.apache.commons.lang.time.FastDateFormat;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.conf.ComponentConfiguration;
+import org.apache.flume.conf.Configurable;
+import org.apache.flume.conf.ConfigurableComponent;
+import org.apache.flume.formatter.output.BucketPath;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.client.Client;
+
+import com.google.common.annotations.VisibleForTesting;
+
+/**
+ * Abstract base class for custom implementations of
+ * {@link ElasticSearchIndexRequestBuilderFactory}.
+ */
+public abstract class AbstractElasticSearchIndexRequestBuilderFactory
+ implements ElasticSearchIndexRequestBuilderFactory {
+
+ /**
+ * {@link FastDateFormat} to use for index names
+ * in {@link #getIndexName(String, long)}
+ */
+ protected final FastDateFormat fastDateFormat;
+
+ /**
+ * Constructor for subclasses
+ * @param fastDateFormat {@link FastDateFormat} to use for index names
+ */
+ protected AbstractElasticSearchIndexRequestBuilderFactory(FastDateFormat fastDateFormat) {
+ this.fastDateFormat = fastDateFormat;
+ }
+
+ /**
+ * @see Configurable
+ */
+ @Override
+ public abstract void configure(Context arg0);
+
+ /**
+ * @see ConfigurableComponent
+ */
+ @Override
+ public abstract void configure(ComponentConfiguration arg0);
+
+ /**
+ * Creates and prepares an {@link IndexRequestBuilder} from the supplied
+ * {@link Client} via delegation to the subclass-hook template methods
+ * {@link #getIndexName(String, long)} and
+ * {@link #prepareIndexRequest(IndexRequestBuilder, String, String, Event)}
+ */
+ @Override
+ public IndexRequestBuilder createIndexRequest(Client client,
+ String indexPrefix, String indexType, Event event) throws IOException {
+ IndexRequestBuilder request = prepareIndex(client);
+ String realIndexPrefix = BucketPath.escapeString(indexPrefix, event.getHeaders());
+ String realIndexType = BucketPath.escapeString(indexType, event.getHeaders());
+
+ TimestampedEvent timestampedEvent = new TimestampedEvent(event);
+ long timestamp = timestampedEvent.getTimestamp();
+
+ String indexName = getIndexName(realIndexPrefix, timestamp);
+ prepareIndexRequest(request, indexName, realIndexType, timestampedEvent);
+ return request;
+ }
+
+ @VisibleForTesting
+ IndexRequestBuilder prepareIndex(Client client) {
+ return client.prepareIndex();
+ }
+
+ /**
+ * Gets the name of the index to use for an index request
+ * @param indexPrefix
+ * Prefix of index name to use -- as configured on the sink
+ * @param timestamp
+ * timestamp (millis) to format / use
+ * @return index name of the form 'indexPrefix-formattedTimestamp'
+ */
+ protected String getIndexName(String indexPrefix, long timestamp) {
+ return new StringBuilder(indexPrefix).append('-')
+ .append(fastDateFormat.format(timestamp)).toString();
+ }
+
+ /**
+ * Prepares an ElasticSearch {@link IndexRequestBuilder} instance
+ * @param indexRequest
+ * The (empty) ElasticSearch {@link IndexRequestBuilder} to prepare
+ * @param indexName
+ * Index name to use -- as per {@link #getIndexName(String, long)}
+ * @param indexType
+ * Index type to use -- as configured on the sink
+ * @param event
+ * Flume event to serialize and add to index request
+ * @throws IOException
+ * If an error occurs e.g. during serialization
+ */
+ protected abstract void prepareIndexRequest(
+ IndexRequestBuilder indexRequest, String indexName,
+ String indexType, Event event) throws IOException;
+
+}
\ No newline at end of file
diff --git a/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/ContentBuilderUtil.java b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/ContentBuilderUtil.java
new file mode 100644
index 0000000..4fda1b8
--- /dev/null
+++ b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/ContentBuilderUtil.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch;
+
+import java.io.IOException;
+import java.nio.charset.Charset;
+
+import org.elasticsearch.common.jackson.core.JsonParseException;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
+import org.elasticsearch.common.xcontent.XContentParser;
+import org.elasticsearch.common.xcontent.XContentType;
+
+/**
+ * Utility methods for using ElasticSearch {@link XContentBuilder}
+ */
+public class ContentBuilderUtil {
+
+ private static final Charset charset = Charset.defaultCharset();
+
+ private ContentBuilderUtil() {
+ }
+
+ public static void appendField(XContentBuilder builder, String field,
+ byte[] data) throws IOException {
+ XContentType contentType = XContentFactory.xContentType(data);
+ if (contentType == null) {
+ addSimpleField(builder, field, data);
+ } else {
+ addComplexField(builder, field, contentType, data);
+ }
+ }
+
+ public static void addSimpleField(XContentBuilder builder, String fieldName,
+ byte[] data) throws IOException {
+ builder.field(fieldName, new String(data, charset));
+ }
+
+ public static void addComplexField(XContentBuilder builder, String fieldName,
+ XContentType contentType, byte[] data) throws IOException {
+ XContentParser parser = null;
+ try {
+ // Elasticsearch will accept JSON directly but we need to validate that
+ // the incoming event is JSON first. Sadly, the elasticsearch JSON parser
+ // is a stream parser so we need to instantiate it, parse the event to
+ // validate it, then instantiate it again to provide the JSON to
+ // elasticsearch.
+ // If validation fails then the incoming event is submitted to
+ // elasticsearch as plain text.
+ parser = XContentFactory.xContent(contentType).createParser(data);
+ while (parser.nextToken() != null) {};
+
+ // If the JSON is valid then include it
+ parser = XContentFactory.xContent(contentType).createParser(data);
+ // Add the field name, but not the value.
+ builder.field(fieldName);
+ // This will add the whole parsed content as the value of the field.
+ builder.copyCurrentStructure(parser);
+ } catch (JsonParseException ex) {
+ // If we get an exception here the most likely cause is nested JSON that
+ // can't be figured out in the body. At this point just push it through
+ // as is
+ addSimpleField(builder, fieldName, data);
+ } finally {
+ if (parser != null) {
+ parser.close();
+ }
+ }
+ }
+}
diff --git a/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/ElasticSearchDynamicSerializer.java b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/ElasticSearchDynamicSerializer.java
new file mode 100644
index 0000000..aa7ad39
--- /dev/null
+++ b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/ElasticSearchDynamicSerializer.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch;
+
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+
+import java.io.IOException;
+import java.util.Map;
+
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.conf.ComponentConfiguration;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+
+/**
+ * Basic serializer that serializes the event body and header fields into
+ * individual fields</p>
+ *
+ * A best effort will be used to determine the content-type, if it cannot be
+ * determined fields will be indexed as Strings
+ */
+public class ElasticSearchDynamicSerializer implements
+ ElasticSearchEventSerializer {
+
+ @Override
+ public void configure(Context context) {
+ // NO-OP...
+ }
+
+ @Override
+ public void configure(ComponentConfiguration conf) {
+ // NO-OP...
+ }
+
+ @Override
+ public XContentBuilder getContentBuilder(Event event) throws IOException {
+ XContentBuilder builder = jsonBuilder().startObject();
+ appendBody(builder, event);
+ appendHeaders(builder, event);
+ return builder;
+ }
+
+ private void appendBody(XContentBuilder builder, Event event)
+ throws IOException {
+ ContentBuilderUtil.appendField(builder, "body", event.getBody());
+ }
+
+ private void appendHeaders(XContentBuilder builder, Event event)
+ throws IOException {
+ Map<String, String> headers = event.getHeaders();
+ for (String key : headers.keySet()) {
+ ContentBuilderUtil.appendField(builder, key,
+ headers.get(key).getBytes(charset));
+ }
+ }
+
+}
diff --git a/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/ElasticSearchEventSerializer.java b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/ElasticSearchEventSerializer.java
new file mode 100644
index 0000000..c89d627
--- /dev/null
+++ b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/ElasticSearchEventSerializer.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch;
+
+import java.io.IOException;
+import java.nio.charset.Charset;
+
+import org.apache.flume.Event;
+import org.apache.flume.conf.Configurable;
+import org.apache.flume.conf.ConfigurableComponent;
+import org.elasticsearch.common.io.BytesStream;
+
+/**
+ * Interface for an event serializer which serializes the headers and body of an
+ * event to write them to ElasticSearch. This is configurable, so any config
+ * params required should be taken through this.
+ */
+public interface ElasticSearchEventSerializer extends Configurable,
+ ConfigurableComponent {
+
+ public static final Charset charset = Charset.defaultCharset();
+
+ /**
+ * Return an {@link BytesStream} made up of the serialized flume event
+ * @param event
+ * The flume event to serialize
+ * @return A {@link BytesStream} used to write to ElasticSearch
+ * @throws IOException
+ * If an error occurs during serialization
+ */
+ abstract BytesStream getContentBuilder(Event event) throws IOException;
+}
diff --git a/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/ElasticSearchIndexRequestBuilderFactory.java b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/ElasticSearchIndexRequestBuilderFactory.java
new file mode 100644
index 0000000..f76308c
--- /dev/null
+++ b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/ElasticSearchIndexRequestBuilderFactory.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch;
+
+import org.apache.commons.lang.time.FastDateFormat;
+import org.apache.flume.Event;
+import org.apache.flume.conf.Configurable;
+import org.apache.flume.conf.ConfigurableComponent;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.client.Client;
+
+import java.io.IOException;
+import java.util.TimeZone;
+
+/**
+ * Interface for creating ElasticSearch {@link IndexRequestBuilder} instances
+ * from serialized flume events. This is configurable, so any config params
+ * required should be taken through this.
+ */
+public interface ElasticSearchIndexRequestBuilderFactory extends Configurable,
+ ConfigurableComponent {
+
+ static final FastDateFormat df = FastDateFormat.getInstance("yyyy-MM-dd",
+ TimeZone.getTimeZone("Etc/UTC"));
+
+ /**
+ * @param client
+ * ElasticSearch {@link Client} to prepare index from
+ * @param indexPrefix
+ * Prefix of index name to use -- as configured on the sink
+ * @param indexType
+ * Index type to use -- as configured on the sink
+ * @param event
+ * Flume event to serialize and add to index request
+ * @return prepared ElasticSearch {@link IndexRequestBuilder} instance
+ * @throws IOException
+ * If an error occurs e.g. during serialization
+ */
+ IndexRequestBuilder createIndexRequest(Client client, String indexPrefix,
+ String indexType, Event event) throws IOException;
+
+
+
+}
diff --git a/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/ElasticSearchLogStashEventSerializer.java b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/ElasticSearchLogStashEventSerializer.java
new file mode 100644
index 0000000..3638368
--- /dev/null
+++ b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/ElasticSearchLogStashEventSerializer.java
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch;
+
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+
+import java.io.IOException;
+import java.io.UnsupportedEncodingException;
+import java.util.Date;
+import java.util.Map;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.conf.ComponentConfiguration;
+import org.elasticsearch.common.collect.Maps;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+
+/**
+ * Serialize flume events into the same format LogStash uses</p>
+ *
+ * This can be used to send events to ElasticSearch and use clients such as
+ * Kabana which expect Logstash formated indexes
+ *
+ * <pre>
+ * {
+ * "@timestamp": "2010-12-21T21:48:33.309258Z",
+ * "@tags": [ "array", "of", "tags" ],
+ * "@type": "string",
+ * "@source": "source of the event, usually a URL."
+ * "@source_host": ""
+ * "@source_path": ""
+ * "@fields":{
+ * # a set of fields for this event
+ * "user": "jordan",
+ * "command": "shutdown -r":
+ * }
+ * "@message": "the original plain-text message"
+ * }
+ * </pre>
+ *
+ * If the following headers are present, they will map to the above logstash
+ * output as long as the logstash fields are not already present.</p>
+ *
+ * <pre>
+ * timestamp: long -> @timestamp:Date
+ * host: String -> @source_host: String
+ * src_path: String -> @source_path: String
+ * type: String -> @type: String
+ * source: String -> @source: String
+ * </pre>
+ *
+ * @see https
+ * ://github.com/logstash/logstash/wiki/logstash%27s-internal-message-
+ * format
+ */
+public class ElasticSearchLogStashEventSerializer implements
+ ElasticSearchEventSerializer {
+
+ @Override
+ public XContentBuilder getContentBuilder(Event event) throws IOException {
+ XContentBuilder builder = jsonBuilder().startObject();
+ appendBody(builder, event);
+ appendHeaders(builder, event);
+ return builder;
+ }
+
+ private void appendBody(XContentBuilder builder, Event event)
+ throws IOException, UnsupportedEncodingException {
+ byte[] body = event.getBody();
+ ContentBuilderUtil.appendField(builder, "@message", body);
+ }
+
+ private void appendHeaders(XContentBuilder builder, Event event)
+ throws IOException {
+ Map<String, String> headers = Maps.newHashMap(event.getHeaders());
+
+ String timestamp = headers.get("timestamp");
+ if (!StringUtils.isBlank(timestamp)
+ && StringUtils.isBlank(headers.get("@timestamp"))) {
+ long timestampMs = Long.parseLong(timestamp);
+ builder.field("@timestamp", new Date(timestampMs));
+ }
+
+ String source = headers.get("source");
+ if (!StringUtils.isBlank(source)
+ && StringUtils.isBlank(headers.get("@source"))) {
+ ContentBuilderUtil.appendField(builder, "@source",
+ source.getBytes(charset));
+ }
+
+ String type = headers.get("type");
+ if (!StringUtils.isBlank(type)
+ && StringUtils.isBlank(headers.get("@type"))) {
+ ContentBuilderUtil.appendField(builder, "@type", type.getBytes(charset));
+ }
+
+ String host = headers.get("host");
+ if (!StringUtils.isBlank(host)
+ && StringUtils.isBlank(headers.get("@source_host"))) {
+ ContentBuilderUtil.appendField(builder, "@source_host",
+ host.getBytes(charset));
+ }
+
+ String srcPath = headers.get("src_path");
+ if (!StringUtils.isBlank(srcPath)
+ && StringUtils.isBlank(headers.get("@source_path"))) {
+ ContentBuilderUtil.appendField(builder, "@source_path",
+ srcPath.getBytes(charset));
+ }
+
+ builder.startObject("@fields");
+ for (String key : headers.keySet()) {
+ byte[] val = headers.get(key).getBytes(charset);
+ ContentBuilderUtil.appendField(builder, key, val);
+ }
+ builder.endObject();
+ }
+
+ @Override
+ public void configure(Context context) {
+ // NO-OP...
+ }
+
+ @Override
+ public void configure(ComponentConfiguration conf) {
+ // NO-OP...
+ }
+}
diff --git a/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/ElasticSearchSink.java b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/ElasticSearchSink.java
new file mode 100644
index 0000000..05eb5ff
--- /dev/null
+++ b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/ElasticSearchSink.java
@@ -0,0 +1,434 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch;
+
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.BATCH_SIZE;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.CLUSTER_NAME;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.DEFAULT_CLUSTER_NAME;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.DEFAULT_INDEX_NAME;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.DEFAULT_INDEX_TYPE;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.DEFAULT_TTL;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.HOSTNAMES;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.INDEX_NAME;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.INDEX_TYPE;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.SERIALIZER;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.SERIALIZER_PREFIX;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.TTL;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.TTL_REGEX;
+import org.apache.commons.lang.StringUtils;
+import org.apache.flume.Channel;
+import org.apache.flume.Context;
+import org.apache.flume.CounterGroup;
+import org.apache.flume.Event;
+import org.apache.flume.EventDeliveryException;
+import org.apache.flume.Transaction;
+import org.apache.flume.conf.BatchSizeSupported;
+import org.apache.flume.formatter.output.BucketPath;
+import org.apache.flume.conf.Configurable;
+import org.apache.flume.instrumentation.SinkCounter;
+import org.apache.flume.sink.AbstractSink;
+import org.apache.flume.sink.elasticsearch.client.ElasticSearchClient;
+import org.apache.flume.sink.elasticsearch.client.ElasticSearchClientFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Throwables;
+
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.CLIENT_PREFIX;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.CLIENT_TYPE;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.DEFAULT_CLIENT_TYPE;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.DEFAULT_INDEX_NAME_BUILDER_CLASS;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.DEFAULT_SERIALIZER_CLASS;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.INDEX_NAME_BUILDER;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.INDEX_NAME_BUILDER_PREFIX;
+
+/**
+ * A sink which reads events from a channel and writes them to ElasticSearch
+ * based on the work done by https://github.com/Aconex/elasticflume.git.</p>
+ *
+ * This sink supports batch reading of events from the channel and writing them
+ * to ElasticSearch.</p>
+ *
+ * Indexes will be rolled daily using the format 'indexname-YYYY-MM-dd' to allow
+ * easier management of the index</p>
+ *
+ * This sink must be configured with with mandatory parameters detailed in
+ * {@link ElasticSearchSinkConstants}</p> It is recommended as a secondary step
+ * the ElasticSearch indexes are optimized for the specified serializer. This is
+ * not handled by the sink but is typically done by deploying a config template
+ * alongside the ElasticSearch deploy</p>
+ *
+ * @see http
+ * ://www.elasticsearch.org/guide/reference/api/admin-indices-templates.
+ * html
+ */
+public class ElasticSearchSink extends AbstractSink implements Configurable, BatchSizeSupported {
+
+ private static final Logger logger = LoggerFactory
+ .getLogger(ElasticSearchSink.class);
+
+ // Used for testing
+ private boolean isLocal = false;
+ private final CounterGroup counterGroup = new CounterGroup();
+
+ private static final int defaultBatchSize = 100;
+
+ private int batchSize = defaultBatchSize;
+ private long ttlMs = DEFAULT_TTL;
+ private String clusterName = DEFAULT_CLUSTER_NAME;
+ private String indexName = DEFAULT_INDEX_NAME;
+ private String indexType = DEFAULT_INDEX_TYPE;
+ private String clientType = DEFAULT_CLIENT_TYPE;
+ private final Pattern pattern = Pattern.compile(TTL_REGEX,
+ Pattern.CASE_INSENSITIVE);
+ private Matcher matcher = pattern.matcher("");
+
+ private String[] serverAddresses = null;
+
+ private ElasticSearchClient client = null;
+ private Context elasticSearchClientContext = null;
+
+ private ElasticSearchIndexRequestBuilderFactory indexRequestFactory;
+ private ElasticSearchEventSerializer eventSerializer;
+ private IndexNameBuilder indexNameBuilder;
+ private SinkCounter sinkCounter;
+
+ /**
+ * Create an {@link ElasticSearchSink} configured using the supplied
+ * configuration
+ */
+ public ElasticSearchSink() {
+ this(false);
+ }
+
+ /**
+ * Create an {@link ElasticSearchSink}</p>
+ *
+ * @param isLocal
+ * If <tt>true</tt> sink will be configured to only talk to an
+ * ElasticSearch instance hosted in the same JVM, should always be
+ * false is production
+ *
+ */
+ @VisibleForTesting
+ ElasticSearchSink(boolean isLocal) {
+ this.isLocal = isLocal;
+ }
+
+ @VisibleForTesting
+ String[] getServerAddresses() {
+ return serverAddresses;
+ }
+
+ @VisibleForTesting
+ String getClusterName() {
+ return clusterName;
+ }
+
+ @VisibleForTesting
+ String getIndexName() {
+ return indexName;
+ }
+
+ @VisibleForTesting
+ String getIndexType() {
+ return indexType;
+ }
+
+ @VisibleForTesting
+ long getTTLMs() {
+ return ttlMs;
+ }
+
+ @VisibleForTesting
+ ElasticSearchEventSerializer getEventSerializer() {
+ return eventSerializer;
+ }
+
+ @VisibleForTesting
+ IndexNameBuilder getIndexNameBuilder() {
+ return indexNameBuilder;
+ }
+
+ @Override
+ public long getBatchSize() {
+ return batchSize;
+ }
+
+ @Override
+ public Status process() throws EventDeliveryException {
+ logger.debug("processing...");
+ Status status = Status.READY;
+ Channel channel = getChannel();
+ Transaction txn = channel.getTransaction();
+ try {
+ txn.begin();
+ int count;
+ for (count = 0; count < batchSize; ++count) {
+ Event event = channel.take();
+
+ if (event == null) {
+ break;
+ }
+ String realIndexType = BucketPath.escapeString(indexType, event.getHeaders());
+ client.addEvent(event, indexNameBuilder, realIndexType, ttlMs);
+ }
+
+ if (count <= 0) {
+ sinkCounter.incrementBatchEmptyCount();
+ counterGroup.incrementAndGet("channel.underflow");
+ status = Status.BACKOFF;
+ } else {
+ if (count < batchSize) {
+ sinkCounter.incrementBatchUnderflowCount();
+ status = Status.BACKOFF;
+ } else {
+ sinkCounter.incrementBatchCompleteCount();
+ }
+
+ sinkCounter.addToEventDrainAttemptCount(count);
+ client.execute();
+ }
+ txn.commit();
+ sinkCounter.addToEventDrainSuccessCount(count);
+ counterGroup.incrementAndGet("transaction.success");
+ } catch (Throwable ex) {
+ try {
+ txn.rollback();
+ counterGroup.incrementAndGet("transaction.rollback");
+ } catch (Exception ex2) {
+ logger.error(
+ "Exception in rollback. Rollback might not have been successful.",
+ ex2);
+ }
+
+ if (ex instanceof Error || ex instanceof RuntimeException) {
+ logger.error("Failed to commit transaction. Transaction rolled back.",
+ ex);
+ Throwables.propagate(ex);
+ } else {
+ logger.error("Failed to commit transaction. Transaction rolled back.",
+ ex);
+ throw new EventDeliveryException(
+ "Failed to commit transaction. Transaction rolled back.", ex);
+ }
+ } finally {
+ txn.close();
+ }
+ return status;
+ }
+
+ @Override
+ public void configure(Context context) {
+ if (!isLocal) {
+ if (StringUtils.isNotBlank(context.getString(HOSTNAMES))) {
+ serverAddresses = StringUtils.deleteWhitespace(
+ context.getString(HOSTNAMES)).split(",");
+ }
+ Preconditions.checkState(serverAddresses != null
+ && serverAddresses.length > 0, "Missing Param:" + HOSTNAMES);
+ }
+
+ if (StringUtils.isNotBlank(context.getString(INDEX_NAME))) {
+ this.indexName = context.getString(INDEX_NAME);
+ }
+
+ if (StringUtils.isNotBlank(context.getString(INDEX_TYPE))) {
+ this.indexType = context.getString(INDEX_TYPE);
+ }
+
+ if (StringUtils.isNotBlank(context.getString(CLUSTER_NAME))) {
+ this.clusterName = context.getString(CLUSTER_NAME);
+ }
+
+ if (StringUtils.isNotBlank(context.getString(BATCH_SIZE))) {
+ this.batchSize = Integer.parseInt(context.getString(BATCH_SIZE));
+ }
+
+ if (StringUtils.isNotBlank(context.getString(TTL))) {
+ this.ttlMs = parseTTL(context.getString(TTL));
+ Preconditions.checkState(ttlMs > 0, TTL
+ + " must be greater than 0 or not set.");
+ }
+
+ if (StringUtils.isNotBlank(context.getString(CLIENT_TYPE))) {
+ clientType = context.getString(CLIENT_TYPE);
+ }
+
+ elasticSearchClientContext = new Context();
+ elasticSearchClientContext.putAll(context.getSubProperties(CLIENT_PREFIX));
+
+ String serializerClazz = DEFAULT_SERIALIZER_CLASS;
+ if (StringUtils.isNotBlank(context.getString(SERIALIZER))) {
+ serializerClazz = context.getString(SERIALIZER);
+ }
+
+ Context serializerContext = new Context();
+ serializerContext.putAll(context.getSubProperties(SERIALIZER_PREFIX));
+
+ try {
+ @SuppressWarnings("unchecked")
+ Class<? extends Configurable> clazz = (Class<? extends Configurable>) Class
+ .forName(serializerClazz);
+ Configurable serializer = clazz.newInstance();
+
+ if (serializer instanceof ElasticSearchIndexRequestBuilderFactory) {
+ indexRequestFactory
+ = (ElasticSearchIndexRequestBuilderFactory) serializer;
+ indexRequestFactory.configure(serializerContext);
+ } else if (serializer instanceof ElasticSearchEventSerializer) {
+ eventSerializer = (ElasticSearchEventSerializer) serializer;
+ eventSerializer.configure(serializerContext);
+ } else {
+ throw new IllegalArgumentException(serializerClazz
+ + " is not an ElasticSearchEventSerializer");
+ }
+ } catch (Exception e) {
+ logger.error("Could not instantiate event serializer.", e);
+ Throwables.propagate(e);
+ }
+
+ if (sinkCounter == null) {
+ sinkCounter = new SinkCounter(getName());
+ }
+
+ String indexNameBuilderClass = DEFAULT_INDEX_NAME_BUILDER_CLASS;
+ if (StringUtils.isNotBlank(context.getString(INDEX_NAME_BUILDER))) {
+ indexNameBuilderClass = context.getString(INDEX_NAME_BUILDER);
+ }
+
+ Context indexnameBuilderContext = new Context();
+ serializerContext.putAll(
+ context.getSubProperties(INDEX_NAME_BUILDER_PREFIX));
+
+ try {
+ @SuppressWarnings("unchecked")
+ Class<? extends IndexNameBuilder> clazz
+ = (Class<? extends IndexNameBuilder>) Class
+ .forName(indexNameBuilderClass);
+ indexNameBuilder = clazz.newInstance();
+ indexnameBuilderContext.put(INDEX_NAME, indexName);
+ indexNameBuilder.configure(indexnameBuilderContext);
+ } catch (Exception e) {
+ logger.error("Could not instantiate index name builder.", e);
+ Throwables.propagate(e);
+ }
+
+ if (sinkCounter == null) {
+ sinkCounter = new SinkCounter(getName());
+ }
+
+ Preconditions.checkState(StringUtils.isNotBlank(indexName),
+ "Missing Param:" + INDEX_NAME);
+ Preconditions.checkState(StringUtils.isNotBlank(indexType),
+ "Missing Param:" + INDEX_TYPE);
+ Preconditions.checkState(StringUtils.isNotBlank(clusterName),
+ "Missing Param:" + CLUSTER_NAME);
+ Preconditions.checkState(batchSize >= 1, BATCH_SIZE
+ + " must be greater than 0");
+ }
+
+ @Override
+ public void start() {
+ ElasticSearchClientFactory clientFactory = new ElasticSearchClientFactory();
+
+ logger.info("ElasticSearch sink {} started");
+ sinkCounter.start();
+ try {
+ if (isLocal) {
+ client = clientFactory.getLocalClient(
+ clientType, eventSerializer, indexRequestFactory);
+ } else {
+ client = clientFactory.getClient(clientType, serverAddresses,
+ clusterName, eventSerializer, indexRequestFactory);
+ client.configure(elasticSearchClientContext);
+ }
+ sinkCounter.incrementConnectionCreatedCount();
+ } catch (Exception ex) {
+ ex.printStackTrace();
+ sinkCounter.incrementConnectionFailedCount();
+ if (client != null) {
+ client.close();
+ sinkCounter.incrementConnectionClosedCount();
+ }
+ }
+
+ super.start();
+ }
+
+ @Override
+ public void stop() {
+ logger.info("ElasticSearch sink {} stopping");
+ if (client != null) {
+ client.close();
+ }
+ sinkCounter.incrementConnectionClosedCount();
+ sinkCounter.stop();
+ super.stop();
+ }
+
+ /*
+ * Returns TTL value of ElasticSearch index in milliseconds when TTL specifier
+ * is "ms" / "s" / "m" / "h" / "d" / "w". In case of unknown specifier TTL is
+ * not set. When specifier is not provided it defaults to days in milliseconds
+ * where the number of days is parsed integer from TTL string provided by
+ * user. <p> Elasticsearch supports ttl values being provided in the format:
+ * 1d / 1w / 1ms / 1s / 1h / 1m specify a time unit like d (days), m
+ * (minutes), h (hours), ms (milliseconds) or w (weeks), milliseconds is used
+ * as default unit.
+ * http://www.elasticsearch.org/guide/reference/mapping/ttl-field/.
+ *
+ * @param ttl TTL value provided by user in flume configuration file for the
+ * sink
+ *
+ * @return the ttl value in milliseconds
+ */
+ private long parseTTL(String ttl) {
+ matcher = matcher.reset(ttl);
+ while (matcher.find()) {
+ if (matcher.group(2).equals("ms")) {
+ return Long.parseLong(matcher.group(1));
+ } else if (matcher.group(2).equals("s")) {
+ return TimeUnit.SECONDS.toMillis(Integer.parseInt(matcher.group(1)));
+ } else if (matcher.group(2).equals("m")) {
+ return TimeUnit.MINUTES.toMillis(Integer.parseInt(matcher.group(1)));
+ } else if (matcher.group(2).equals("h")) {
+ return TimeUnit.HOURS.toMillis(Integer.parseInt(matcher.group(1)));
+ } else if (matcher.group(2).equals("d")) {
+ return TimeUnit.DAYS.toMillis(Integer.parseInt(matcher.group(1)));
+ } else if (matcher.group(2).equals("w")) {
+ return TimeUnit.DAYS.toMillis(7 * Integer.parseInt(matcher.group(1)));
+ } else if (matcher.group(2).equals("")) {
+ logger.info("TTL qualifier is empty. Defaulting to day qualifier.");
+ return TimeUnit.DAYS.toMillis(Integer.parseInt(matcher.group(1)));
+ } else {
+ logger.debug("Unknown TTL qualifier provided. Setting TTL to 0.");
+ return 0;
+ }
+ }
+ logger.info("TTL not provided. Skipping the TTL config by returning 0.");
+ return 0;
+ }
+}
diff --git a/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/ElasticSearchSinkConstants.java b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/ElasticSearchSinkConstants.java
new file mode 100644
index 0000000..da88def
--- /dev/null
+++ b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/ElasticSearchSinkConstants.java
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch;
+
+public class ElasticSearchSinkConstants {
+
+ /**
+ * Comma separated list of hostname:port, if the port is not present the
+ * default port '9300' will be used</p>
+ * Example:
+ * <pre>
+ * 127.0.0.1:92001,127.0.0.2:9300
+ * </pre>
+ */
+ public static final String HOSTNAMES = "hostNames";
+
+ /**
+ * The name to index the document to, defaults to 'flume'</p>
+ * The current date in the format 'yyyy-MM-dd' will be appended to this name,
+ * for example 'foo' will result in a daily index of 'foo-yyyy-MM-dd'
+ */
+ public static final String INDEX_NAME = "indexName";
+
+ /**
+ * The type to index the document to, defaults to 'log'
+ */
+ public static final String INDEX_TYPE = "indexType";
+
+ /**
+ * Name of the ElasticSearch cluster to connect to
+ */
+ public static final String CLUSTER_NAME = "clusterName";
+
+ /**
+ * Maximum number of events the sink should take from the channel per
+ * transaction, if available. Defaults to 100
+ */
+ public static final String BATCH_SIZE = "batchSize";
+
+ /**
+ * TTL in days, when set will cause the expired documents to be deleted
+ * automatically, if not set documents will never be automatically deleted
+ */
+ public static final String TTL = "ttl";
+
+ /**
+ * The fully qualified class name of the serializer the sink should use.
+ */
+ public static final String SERIALIZER = "serializer";
+
+ /**
+ * Configuration to pass to the serializer.
+ */
+ public static final String SERIALIZER_PREFIX = SERIALIZER + ".";
+
+ /**
+ * The fully qualified class name of the index name builder the sink
+ * should use to determine name of index where the event should be sent.
+ */
+ public static final String INDEX_NAME_BUILDER = "indexNameBuilder";
+
+ /**
+ * The fully qualified class name of the index name builder the sink
+ * should use to determine name of index where the event should be sent.
+ */
+ public static final String INDEX_NAME_BUILDER_PREFIX
+ = INDEX_NAME_BUILDER + ".";
+
+ /**
+ * The client type used for sending bulks to ElasticSearch
+ */
+ public static final String CLIENT_TYPE = "client";
+
+ /**
+ * The client prefix to extract the configuration that will be passed to
+ * elasticsearch client.
+ */
+ public static final String CLIENT_PREFIX = CLIENT_TYPE + ".";
+
+ /**
+ * DEFAULTS USED BY THE SINK
+ */
+
+ public static final int DEFAULT_PORT = 9300;
+ public static final int DEFAULT_TTL = -1;
+ public static final String DEFAULT_INDEX_NAME = "flume";
+ public static final String DEFAULT_INDEX_TYPE = "log";
+ public static final String DEFAULT_CLUSTER_NAME = "elasticsearch";
+ public static final String DEFAULT_CLIENT_TYPE = "transport";
+ public static final String TTL_REGEX = "^(\\d+)(\\D*)";
+ public static final String DEFAULT_SERIALIZER_CLASS = "org.apache.flume." +
+ "sink.elasticsearch.ElasticSearchLogStashEventSerializer";
+ public static final String DEFAULT_INDEX_NAME_BUILDER_CLASS =
+ "org.apache.flume.sink.elasticsearch.TimeBasedIndexNameBuilder";
+}
diff --git a/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/EventSerializerIndexRequestBuilderFactory.java b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/EventSerializerIndexRequestBuilderFactory.java
new file mode 100644
index 0000000..d6cca50
--- /dev/null
+++ b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/EventSerializerIndexRequestBuilderFactory.java
@@ -0,0 +1,69 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch;
+
+import java.io.IOException;
+
+import org.apache.commons.lang.time.FastDateFormat;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.conf.ComponentConfiguration;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.common.io.BytesStream;
+
+/**
+ * Default implementation of {@link ElasticSearchIndexRequestBuilderFactory}.
+ * It serializes flume events using the
+ * {@link ElasticSearchEventSerializer} instance configured on the sink.
+ */
+public class EventSerializerIndexRequestBuilderFactory
+ extends AbstractElasticSearchIndexRequestBuilderFactory {
+
+ protected final ElasticSearchEventSerializer serializer;
+
+ public EventSerializerIndexRequestBuilderFactory(
+ ElasticSearchEventSerializer serializer) {
+ this(serializer, ElasticSearchIndexRequestBuilderFactory.df);
+ }
+
+ protected EventSerializerIndexRequestBuilderFactory(
+ ElasticSearchEventSerializer serializer, FastDateFormat fdf) {
+ super(fdf);
+ this.serializer = serializer;
+ }
+
+ @Override
+ public void configure(Context context) {
+ serializer.configure(context);
+ }
+
+ @Override
+ public void configure(ComponentConfiguration config) {
+ serializer.configure(config);
+ }
+
+ @Override
+ protected void prepareIndexRequest(IndexRequestBuilder indexRequest,
+ String indexName, String indexType, Event event) throws IOException {
+ BytesStream contentBuilder = serializer.getContentBuilder(event);
+ indexRequest.setIndex(indexName)
+ .setType(indexType)
+ .setSource(contentBuilder.bytes());
+ }
+}
diff --git a/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/IndexNameBuilder.java b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/IndexNameBuilder.java
new file mode 100644
index 0000000..1dd4415
--- /dev/null
+++ b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/IndexNameBuilder.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch;
+
+import org.apache.flume.Event;
+import org.apache.flume.conf.Configurable;
+import org.apache.flume.conf.ConfigurableComponent;
+
+public interface IndexNameBuilder extends Configurable,
+ ConfigurableComponent {
+ /**
+ * Gets the name of the index to use for an index request
+ * @param event
+ * Event which determines index name
+ * @return index name of the form 'indexPrefix-indexDynamicName'
+ */
+ public String getIndexName(Event event);
+
+ /**
+ * Gets the prefix of index to use for an index request.
+ * @param event
+ * Event which determines index name
+ * @return Index prefix name
+ */
+ public String getIndexPrefix(Event event);
+}
diff --git a/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/SimpleIndexNameBuilder.java b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/SimpleIndexNameBuilder.java
new file mode 100644
index 0000000..801cac9
--- /dev/null
+++ b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/SimpleIndexNameBuilder.java
@@ -0,0 +1,46 @@
+/*
+ * Copyright 2014 Apache Software Foundation.
+ *
+ * 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 org.apache.flume.sink.elasticsearch;
+
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.conf.ComponentConfiguration;
+import org.apache.flume.formatter.output.BucketPath;
+
+public class SimpleIndexNameBuilder implements IndexNameBuilder {
+
+ private String indexName;
+
+ @Override
+ public String getIndexName(Event event) {
+ return BucketPath.escapeString(indexName, event.getHeaders());
+ }
+
+ @Override
+ public String getIndexPrefix(Event event) {
+ return BucketPath.escapeString(indexName, event.getHeaders());
+ }
+
+ @Override
+ public void configure(Context context) {
+ indexName = context.getString(ElasticSearchSinkConstants.INDEX_NAME);
+ }
+
+ @Override
+ public void configure(ComponentConfiguration conf) {
+ }
+}
diff --git a/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/TimeBasedIndexNameBuilder.java b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/TimeBasedIndexNameBuilder.java
new file mode 100644
index 0000000..c651732
--- /dev/null
+++ b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/TimeBasedIndexNameBuilder.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.commons.lang.StringUtils;
+import org.apache.commons.lang.time.FastDateFormat;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.conf.ComponentConfiguration;
+import org.apache.flume.formatter.output.BucketPath;
+
+import java.util.TimeZone;
+
+/**
+ * Default index name builder. It prepares name of index using configured
+ * prefix and current timestamp. Default format of name is prefix-yyyy-MM-dd".
+ */
+public class TimeBasedIndexNameBuilder implements
+ IndexNameBuilder {
+
+ public static final String DATE_FORMAT = "dateFormat";
+ public static final String TIME_ZONE = "timeZone";
+
+ public static final String DEFAULT_DATE_FORMAT = "yyyy-MM-dd";
+ public static final String DEFAULT_TIME_ZONE = "Etc/UTC";
+
+ private FastDateFormat fastDateFormat = FastDateFormat.getInstance("yyyy-MM-dd",
+ TimeZone.getTimeZone("Etc/UTC"));
+
+ private String indexPrefix;
+
+ @VisibleForTesting
+ FastDateFormat getFastDateFormat() {
+ return fastDateFormat;
+ }
+
+ /**
+ * Gets the name of the index to use for an index request
+ * @param event
+ * Event for which the name of index has to be prepared
+ * @return index name of the form 'indexPrefix-formattedTimestamp'
+ */
+ @Override
+ public String getIndexName(Event event) {
+ TimestampedEvent timestampedEvent = new TimestampedEvent(event);
+ long timestamp = timestampedEvent.getTimestamp();
+ String realIndexPrefix = BucketPath.escapeString(indexPrefix, event.getHeaders());
+ return new StringBuilder(realIndexPrefix).append('-')
+ .append(fastDateFormat.format(timestamp)).toString();
+ }
+
+ @Override
+ public String getIndexPrefix(Event event) {
+ return BucketPath.escapeString(indexPrefix, event.getHeaders());
+ }
+
+ @Override
+ public void configure(Context context) {
+ String dateFormatString = context.getString(DATE_FORMAT);
+ String timeZoneString = context.getString(TIME_ZONE);
+ if (StringUtils.isBlank(dateFormatString)) {
+ dateFormatString = DEFAULT_DATE_FORMAT;
+ }
+ if (StringUtils.isBlank(timeZoneString)) {
+ timeZoneString = DEFAULT_TIME_ZONE;
+ }
+ fastDateFormat = FastDateFormat.getInstance(dateFormatString,
+ TimeZone.getTimeZone(timeZoneString));
+ indexPrefix = context.getString(ElasticSearchSinkConstants.INDEX_NAME);
+ }
+
+ @Override
+ public void configure(ComponentConfiguration conf) {
+ }
+}
diff --git a/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/TimestampedEvent.java b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/TimestampedEvent.java
new file mode 100644
index 0000000..c056839
--- /dev/null
+++ b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/TimestampedEvent.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch;
+
+import com.google.common.collect.Maps;
+import org.apache.commons.lang.StringUtils;
+import org.apache.flume.Event;
+import org.apache.flume.event.SimpleEvent;
+import org.joda.time.DateTimeUtils;
+
+import java.util.Map;
+
+/**
+ * {@link org.apache.flume.Event} implementation that has a timestamp.
+ * The timestamp is taken from (in order of precedence):<ol>
+ * <li>The "timestamp" header of the base event, if present</li>
+ * <li>The "@timestamp" header of the base event, if present</li>
+ * <li>The current time in millis, otherwise</li>
+ * </ol>
+ */
+final class TimestampedEvent extends SimpleEvent {
+
+ private final long timestamp;
+
+ TimestampedEvent(Event base) {
+ setBody(base.getBody());
+ Map<String, String> headers = Maps.newHashMap(base.getHeaders());
+ String timestampString = headers.get("timestamp");
+ if (StringUtils.isBlank(timestampString)) {
+ timestampString = headers.get("@timestamp");
+ }
+ if (StringUtils.isBlank(timestampString)) {
+ this.timestamp = DateTimeUtils.currentTimeMillis();
+ headers.put("timestamp", String.valueOf(timestamp ));
+ } else {
+ this.timestamp = Long.valueOf(timestampString);
+ }
+ setHeaders(headers);
+ }
+
+ long getTimestamp() {
+ return timestamp;
+ }
+}
diff --git a/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/client/ElasticSearchClient.java b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/client/ElasticSearchClient.java
new file mode 100644
index 0000000..655e00a
--- /dev/null
+++ b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/client/ElasticSearchClient.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch.client;
+
+import org.apache.flume.Event;
+import org.apache.flume.conf.Configurable;
+import org.apache.flume.sink.elasticsearch.IndexNameBuilder;
+
+/**
+ * Interface for an ElasticSearch client which is responsible for sending bulks
+ * of events to ElasticSearch.
+ */
+public interface ElasticSearchClient extends Configurable {
+
+ /**
+ * Close connection to elastic search in client
+ */
+ void close();
+
+ /**
+ * Add new event to the bulk
+ *
+ * @param event
+ * Flume Event
+ * @param indexNameBuilder
+ * Index name builder which generates name of index to feed
+ * @param indexType
+ * Name of type of document which will be sent to the elasticsearch cluster
+ * @param ttlMs
+ * Time to live expressed in milliseconds. Value <= 0 is ignored
+ * @throws Exception
+ */
+ public void addEvent(Event event, IndexNameBuilder indexNameBuilder,
+ String indexType, long ttlMs) throws Exception;
+
+ /**
+ * Sends bulk to the elasticsearch cluster
+ *
+ * @throws Exception
+ */
+ void execute() throws Exception;
+}
diff --git a/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/client/ElasticSearchClientFactory.java b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/client/ElasticSearchClientFactory.java
new file mode 100644
index 0000000..986fb2b
--- /dev/null
+++ b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/client/ElasticSearchClientFactory.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch.client;
+
+import org.apache.flume.sink.elasticsearch.ElasticSearchEventSerializer;
+import org.apache.flume.sink.elasticsearch.ElasticSearchIndexRequestBuilderFactory;
+
+/**
+ * Internal ElasticSearch client factory. Responsible for creating instance
+ * of ElasticSearch clients.
+ */
+public class ElasticSearchClientFactory {
+ public static final String TransportClient = "transport";
+ public static final String RestClient = "rest";
+
+ /**
+ *
+ * @param clientType
+ * String representation of client type
+ * @param hostNames
+ * Array of strings that represents hostnames with ports (hostname:port)
+ * @param clusterName
+ * Elasticsearch cluster name used only by Transport Client
+ * @param serializer
+ * Serializer of flume events to elasticsearch documents
+ * @return
+ */
+ public ElasticSearchClient getClient(String clientType, String[] hostNames,
+ String clusterName, ElasticSearchEventSerializer serializer,
+ ElasticSearchIndexRequestBuilderFactory indexBuilder) throws NoSuchClientTypeException {
+ if (clientType.equalsIgnoreCase(TransportClient) && serializer != null) {
+ return new ElasticSearchTransportClient(hostNames, clusterName, serializer);
+ } else if (clientType.equalsIgnoreCase(TransportClient) && indexBuilder != null) {
+ return new ElasticSearchTransportClient(hostNames, clusterName, indexBuilder);
+ } else if (clientType.equalsIgnoreCase(RestClient) && serializer != null) {
+ return new ElasticSearchRestClient(hostNames, serializer);
+ }
+ throw new NoSuchClientTypeException();
+ }
+
+ /**
+ * Used for tests only. Creates local elasticsearch instance client.
+ *
+ * @param clientType Name of client to use
+ * @param serializer Serializer for the event
+ * @param indexBuilder Index builder factory
+ *
+ * @return Local elastic search instance client
+ */
+ public ElasticSearchClient getLocalClient(String clientType,
+ ElasticSearchEventSerializer serializer,
+ ElasticSearchIndexRequestBuilderFactory indexBuilder)
+ throws NoSuchClientTypeException {
+ if (clientType.equalsIgnoreCase(TransportClient) && serializer != null) {
+ return new ElasticSearchTransportClient(serializer);
+ } else if (clientType.equalsIgnoreCase(TransportClient) && indexBuilder != null) {
+ return new ElasticSearchTransportClient(indexBuilder);
+ } else if (clientType.equalsIgnoreCase(RestClient)) {
+ }
+ throw new NoSuchClientTypeException();
+ }
+}
diff --git a/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/client/ElasticSearchRestClient.java b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/client/ElasticSearchRestClient.java
new file mode 100644
index 0000000..e51efe2
--- /dev/null
+++ b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/client/ElasticSearchRestClient.java
@@ -0,0 +1,149 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch.client;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.gson.Gson;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.EventDeliveryException;
+import org.apache.flume.sink.elasticsearch.ElasticSearchEventSerializer;
+import org.apache.flume.sink.elasticsearch.IndexNameBuilder;
+import org.apache.http.HttpResponse;
+import org.apache.http.HttpStatus;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.DefaultHttpClient;
+import org.apache.http.util.EntityUtils;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Rest ElasticSearch client which is responsible for sending bulks of events to
+ * ElasticSearch using ElasticSearch HTTP API. This is configurable, so any
+ * config params required should be taken through this.
+ */
+public class ElasticSearchRestClient implements ElasticSearchClient {
+
+ private static final String INDEX_OPERATION_NAME = "index";
+ private static final String INDEX_PARAM = "_index";
+ private static final String TYPE_PARAM = "_type";
+ private static final String TTL_PARAM = "_ttl";
+ private static final String BULK_ENDPOINT = "_bulk";
+
+ private static final Logger logger = LoggerFactory.getLogger(ElasticSearchRestClient.class);
+
+ private final ElasticSearchEventSerializer serializer;
+ private final RoundRobinList<String> serversList;
+
+ private StringBuilder bulkBuilder;
+ private HttpClient httpClient;
+
+ public ElasticSearchRestClient(String[] hostNames,
+ ElasticSearchEventSerializer serializer) {
+
+ for (int i = 0; i < hostNames.length; ++i) {
+ if (!hostNames[i].contains("http://") && !hostNames[i].contains("https://")) {
+ hostNames[i] = "http://" + hostNames[i];
+ }
+ }
+ this.serializer = serializer;
+
+ serversList = new RoundRobinList<String>(Arrays.asList(hostNames));
+ httpClient = new DefaultHttpClient();
+ bulkBuilder = new StringBuilder();
+ }
+
+ @VisibleForTesting
+ public ElasticSearchRestClient(String[] hostNames,
+ ElasticSearchEventSerializer serializer, HttpClient client) {
+ this(hostNames, serializer);
+ httpClient = client;
+ }
+
+ @Override
+ public void configure(Context context) {
+ }
+
+ @Override
+ public void close() {
+ }
+
+ @Override
+ public void addEvent(Event event, IndexNameBuilder indexNameBuilder, String indexType,
+ long ttlMs) throws Exception {
+ BytesReference content = serializer.getContentBuilder(event).bytes();
+ Map<String, Map<String, String>> parameters = new HashMap<String, Map<String, String>>();
+ Map<String, String> indexParameters = new HashMap<String, String>();
+ indexParameters.put(INDEX_PARAM, indexNameBuilder.getIndexName(event));
+ indexParameters.put(TYPE_PARAM, indexType);
+ if (ttlMs > 0) {
+ indexParameters.put(TTL_PARAM, Long.toString(ttlMs));
+ }
+ parameters.put(INDEX_OPERATION_NAME, indexParameters);
+
+ Gson gson = new Gson();
+ synchronized (bulkBuilder) {
+ bulkBuilder.append(gson.toJson(parameters));
+ bulkBuilder.append("\n");
+ bulkBuilder.append(content.toBytesArray().toUtf8());
+ bulkBuilder.append("\n");
+ }
+ }
+
+ @Override
+ public void execute() throws Exception {
+ int statusCode = 0, triesCount = 0;
+ HttpResponse response = null;
+ String entity;
+ synchronized (bulkBuilder) {
+ entity = bulkBuilder.toString();
+ bulkBuilder = new StringBuilder();
+ }
+
+ while (statusCode != HttpStatus.SC_OK && triesCount < serversList.size()) {
+ triesCount++;
+ String host = serversList.get();
+ String url = host + "/" + BULK_ENDPOINT;
+ HttpPost httpRequest = new HttpPost(url);
+ httpRequest.setEntity(new StringEntity(entity));
+ response = httpClient.execute(httpRequest);
+ statusCode = response.getStatusLine().getStatusCode();
+ logger.info("Status code from elasticsearch: " + statusCode);
+ if (response.getEntity() != null) {
+ logger.debug("Status message from elasticsearch: " +
+ EntityUtils.toString(response.getEntity(), "UTF-8"));
+ }
+ }
+
+ if (statusCode != HttpStatus.SC_OK) {
+ if (response.getEntity() != null) {
+ throw new EventDeliveryException(EntityUtils.toString(response.getEntity(), "UTF-8"));
+ } else {
+ throw new EventDeliveryException("Elasticsearch status code was: " + statusCode);
+ }
+ }
+ }
+}
diff --git a/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/client/ElasticSearchTransportClient.java b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/client/ElasticSearchTransportClient.java
new file mode 100644
index 0000000..2cf365e
--- /dev/null
+++ b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/client/ElasticSearchTransportClient.java
@@ -0,0 +1,228 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch.client;
+
+import com.google.common.annotations.VisibleForTesting;
+import java.io.IOException;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.EventDeliveryException;
+import org.apache.flume.sink.elasticsearch.ElasticSearchEventSerializer;
+import org.apache.flume.sink.elasticsearch.IndexNameBuilder;
+import org.elasticsearch.action.bulk.BulkRequestBuilder;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.client.transport.TransportClient;
+import org.elasticsearch.common.settings.ImmutableSettings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.transport.InetSocketTransportAddress;
+import org.elasticsearch.node.Node;
+import org.elasticsearch.node.NodeBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import org.apache.flume.sink.elasticsearch.ElasticSearchIndexRequestBuilderFactory;
+
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.DEFAULT_PORT;
+
+public class ElasticSearchTransportClient implements ElasticSearchClient {
+
+ public static final Logger logger = LoggerFactory
+ .getLogger(ElasticSearchTransportClient.class);
+
+ private InetSocketTransportAddress[] serverAddresses;
+ private ElasticSearchEventSerializer serializer;
+ private ElasticSearchIndexRequestBuilderFactory indexRequestBuilderFactory;
+ private BulkRequestBuilder bulkRequestBuilder;
+
+ private Client client;
+
+ @VisibleForTesting
+ InetSocketTransportAddress[] getServerAddresses() {
+ return serverAddresses;
+ }
+
+ @VisibleForTesting
+ void setBulkRequestBuilder(BulkRequestBuilder bulkRequestBuilder) {
+ this.bulkRequestBuilder = bulkRequestBuilder;
+ }
+
+ /**
+ * Transport client for external cluster
+ *
+ * @param hostNames
+ * @param clusterName
+ * @param serializer
+ */
+ public ElasticSearchTransportClient(String[] hostNames, String clusterName,
+ ElasticSearchEventSerializer serializer) {
+ configureHostnames(hostNames);
+ this.serializer = serializer;
+ openClient(clusterName);
+ }
+
+ public ElasticSearchTransportClient(String[] hostNames, String clusterName,
+ ElasticSearchIndexRequestBuilderFactory indexBuilder) {
+ configureHostnames(hostNames);
+ this.indexRequestBuilderFactory = indexBuilder;
+ openClient(clusterName);
+ }
+
+ /**
+ * Local transport client only for testing
+ *
+ * @param indexBuilderFactory
+ */
+ public ElasticSearchTransportClient(ElasticSearchIndexRequestBuilderFactory indexBuilderFactory) {
+ this.indexRequestBuilderFactory = indexBuilderFactory;
+ openLocalDiscoveryClient();
+ }
+
+ /**
+ * Local transport client only for testing
+ *
+ * @param serializer
+ */
+ public ElasticSearchTransportClient(ElasticSearchEventSerializer serializer) {
+ this.serializer = serializer;
+ openLocalDiscoveryClient();
+ }
+
+ /**
+ * Used for testing
+ *
+ * @param client
+ * ElasticSearch Client
+ * @param serializer
+ * Event Serializer
+ */
+ public ElasticSearchTransportClient(Client client,
+ ElasticSearchEventSerializer serializer) {
+ this.client = client;
+ this.serializer = serializer;
+ }
+
+ /**
+ * Used for testing
+ */
+ public ElasticSearchTransportClient(Client client,
+ ElasticSearchIndexRequestBuilderFactory requestBuilderFactory)
+ throws IOException {
+ this.client = client;
+ requestBuilderFactory.createIndexRequest(client, null, null, null);
+ }
+
+ private void configureHostnames(String[] hostNames) {
+ logger.warn(Arrays.toString(hostNames));
+ serverAddresses = new InetSocketTransportAddress[hostNames.length];
+ for (int i = 0; i < hostNames.length; i++) {
+ String[] hostPort = hostNames[i].trim().split(":");
+ String host = hostPort[0].trim();
+ int port = hostPort.length == 2 ? Integer.parseInt(hostPort[1].trim())
+ : DEFAULT_PORT;
+ serverAddresses[i] = new InetSocketTransportAddress(host, port);
+ }
+ }
+
+ @Override
+ public void close() {
+ if (client != null) {
+ client.close();
+ }
+ client = null;
+ }
+
+ @Override
+ public void addEvent(Event event, IndexNameBuilder indexNameBuilder,
+ String indexType, long ttlMs) throws Exception {
+ if (bulkRequestBuilder == null) {
+ bulkRequestBuilder = client.prepareBulk();
+ }
+
+ IndexRequestBuilder indexRequestBuilder = null;
+ if (indexRequestBuilderFactory == null) {
+ indexRequestBuilder = client
+ .prepareIndex(indexNameBuilder.getIndexName(event), indexType)
+ .setSource(serializer.getContentBuilder(event).bytes());
+ } else {
+ indexRequestBuilder = indexRequestBuilderFactory.createIndexRequest(
+ client, indexNameBuilder.getIndexPrefix(event), indexType, event);
+ }
+
+ if (ttlMs > 0) {
+ indexRequestBuilder.setTTL(ttlMs);
+ }
+ bulkRequestBuilder.add(indexRequestBuilder);
+ }
+
+ @Override
+ public void execute() throws Exception {
+ try {
+ BulkResponse bulkResponse = bulkRequestBuilder.execute().actionGet();
+ if (bulkResponse.hasFailures()) {
+ throw new EventDeliveryException(bulkResponse.buildFailureMessage());
+ }
+ } finally {
+ bulkRequestBuilder = client.prepareBulk();
+ }
+ }
+
+ /**
+ * Open client to elaticsearch cluster
+ *
+ * @param clusterName
+ */
+ private void openClient(String clusterName) {
+ logger.info("Using ElasticSearch hostnames: {} ",
+ Arrays.toString(serverAddresses));
+ Settings settings = ImmutableSettings.settingsBuilder()
+ .put("cluster.name", clusterName).build();
+
+ TransportClient transportClient = new TransportClient(settings);
+ for (InetSocketTransportAddress host : serverAddresses) {
+ transportClient.addTransportAddress(host);
+ }
+ if (client != null) {
+ client.close();
+ }
+ client = transportClient;
+ }
+
+ /*
+ * FOR TESTING ONLY...
+ *
+ * Opens a local discovery node for talking to an elasticsearch server running
+ * in the same JVM
+ */
+ private void openLocalDiscoveryClient() {
+ logger.info("Using ElasticSearch AutoDiscovery mode");
+ Node node = NodeBuilder.nodeBuilder().client(true).local(true).node();
+ if (client != null) {
+ client.close();
+ }
+ client = node.client();
+ }
+
+ @Override
+ public void configure(Context context) {
+ //To change body of implemented methods use File | Settings | File Templates.
+ }
+}
diff --git a/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/client/NoSuchClientTypeException.java b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/client/NoSuchClientTypeException.java
new file mode 100644
index 0000000..41fbe0d
--- /dev/null
+++ b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/client/NoSuchClientTypeException.java
@@ -0,0 +1,23 @@
+/*
+ * Copyright 2014 Apache Software Foundation.
+ *
+ * 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 org.apache.flume.sink.elasticsearch.client;
+
+/**
+ * Exception class
+ */
+class NoSuchClientTypeException extends Exception {
+}
diff --git a/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/client/RoundRobinList.java b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/client/RoundRobinList.java
new file mode 100644
index 0000000..4cbbe91
--- /dev/null
+++ b/flume-elasticsearch-sink/src/main/java/org/apache/flume/sink/elasticsearch/client/RoundRobinList.java
@@ -0,0 +1,44 @@
+package org.apache.flume.sink.elasticsearch.client;
+
+import java.util.Collection;
+import java.util.Iterator;
+
+/*
+ * Copyright 2014 Apache Software Foundation.
+ *
+ * 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.
+ */
+
+public class RoundRobinList<T> {
+
+ private Iterator<T> iterator;
+ private final Collection<T> elements;
+
+ public RoundRobinList(Collection<T> elements) {
+ this.elements = elements;
+ iterator = this.elements.iterator();
+ }
+
+ public synchronized T get() {
+ if (iterator.hasNext()) {
+ return iterator.next();
+ } else {
+ iterator = elements.iterator();
+ return iterator.next();
+ }
+ }
+
+ public int size() {
+ return elements.size();
+ }
+}
diff --git a/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/AbstractElasticSearchSinkTest.java b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/AbstractElasticSearchSinkTest.java
new file mode 100644
index 0000000..9fbd747
--- /dev/null
+++ b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/AbstractElasticSearchSinkTest.java
@@ -0,0 +1,164 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch;
+
+import org.apache.flume.Channel;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.channel.MemoryChannel;
+import org.apache.flume.conf.Configurables;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.common.collect.Maps;
+import org.elasticsearch.common.settings.ImmutableSettings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.gateway.Gateway;
+import org.elasticsearch.index.query.QueryBuilder;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.node.Node;
+import org.elasticsearch.node.NodeBuilder;
+import org.elasticsearch.node.internal.InternalNode;
+import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.search.SearchHits;
+import org.joda.time.DateTimeUtils;
+import org.junit.After;
+import org.junit.Before;
+
+import java.util.Arrays;
+import java.util.Comparator;
+import java.util.Map;
+
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.BATCH_SIZE;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.CLUSTER_NAME;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.INDEX_NAME;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.INDEX_TYPE;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.TTL;
+import static org.junit.Assert.assertEquals;
+
+public abstract class AbstractElasticSearchSinkTest {
+
+ static final String DEFAULT_INDEX_NAME = "flume";
+ static final String DEFAULT_INDEX_TYPE = "log";
+ static final String DEFAULT_CLUSTER_NAME = "elasticsearch";
+ static final long FIXED_TIME_MILLIS = 123456789L;
+
+ Node node;
+ Client client;
+ String timestampedIndexName;
+ Map<String, String> parameters;
+
+ void initDefaults() {
+ parameters = Maps.newHashMap();
+ parameters.put(INDEX_NAME, DEFAULT_INDEX_NAME);
+ parameters.put(INDEX_TYPE, DEFAULT_INDEX_TYPE);
+ parameters.put(CLUSTER_NAME, DEFAULT_CLUSTER_NAME);
+ parameters.put(BATCH_SIZE, "1");
+ parameters.put(TTL, "5");
+
+ timestampedIndexName = DEFAULT_INDEX_NAME + '-'
+ + ElasticSearchIndexRequestBuilderFactory.df.format(FIXED_TIME_MILLIS);
+ }
+
+ void createNodes() throws Exception {
+ Settings settings = ImmutableSettings
+ .settingsBuilder()
+ .put("number_of_shards", 1)
+ .put("number_of_replicas", 0)
+ .put("routing.hash.type", "simple")
+ .put("gateway.type", "none")
+ .put("path.data", "target/es-test")
+ .build();
+
+ node = NodeBuilder.nodeBuilder().settings(settings).local(true).node();
+ client = node.client();
+
+ client.admin().cluster().prepareHealth().setWaitForGreenStatus().execute()
+ .actionGet();
+ }
+
+ void shutdownNodes() throws Exception {
+ ((InternalNode) node).injector().getInstance(Gateway.class).reset();
+ client.close();
+ node.close();
+ }
+
+ @Before
+ public void setFixedJodaTime() {
+ DateTimeUtils.setCurrentMillisFixed(FIXED_TIME_MILLIS);
+ }
+
+ @After
+ public void resetJodaTime() {
+ DateTimeUtils.setCurrentMillisSystem();
+ }
+
+ Channel bindAndStartChannel(ElasticSearchSink fixture) {
+ // Configure the channel
+ Channel channel = new MemoryChannel();
+ Configurables.configure(channel, new Context());
+
+ // Wire them together
+ fixture.setChannel(channel);
+ fixture.start();
+ return channel;
+ }
+
+ void assertMatchAllQuery(int expectedHits, Event... events) {
+ assertSearch(expectedHits, performSearch(QueryBuilders.matchAllQuery()),
+ null, events);
+ }
+
+ void assertBodyQuery(int expectedHits, Event... events) {
+ // Perform Multi Field Match
+ assertSearch(expectedHits,
+ performSearch(QueryBuilders.fieldQuery("@message", "event")),
+ null, events);
+ }
+
+ SearchResponse performSearch(QueryBuilder query) {
+ return client.prepareSearch(timestampedIndexName)
+ .setTypes(DEFAULT_INDEX_TYPE).setQuery(query).execute().actionGet();
+ }
+
+ void assertSearch(int expectedHits, SearchResponse response, Map<String, Object> expectedBody,
+ Event... events) {
+ SearchHits hitResponse = response.getHits();
+ assertEquals(expectedHits, hitResponse.getTotalHits());
+
+ SearchHit[] hits = hitResponse.getHits();
+ Arrays.sort(hits, new Comparator<SearchHit>() {
+ @Override
+ public int compare(SearchHit o1, SearchHit o2) {
+ return o1.getSourceAsString().compareTo(o2.getSourceAsString());
+ }
+ });
+
+ for (int i = 0; i < events.length; i++) {
+ Event event = events[i];
+ SearchHit hit = hits[i];
+ Map<String, Object> source = hit.getSource();
+ if (expectedBody == null) {
+ assertEquals(new String(event.getBody()), source.get("@message"));
+ } else {
+ assertEquals(expectedBody, source.get("@message"));
+ }
+ }
+ }
+
+}
diff --git a/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/TestElasticSearchDynamicSerializer.java b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/TestElasticSearchDynamicSerializer.java
new file mode 100644
index 0000000..d4e4654
--- /dev/null
+++ b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/TestElasticSearchDynamicSerializer.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch;
+
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.event.EventBuilder;
+import org.elasticsearch.common.collect.Maps;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.junit.Test;
+
+import java.util.Map;
+
+import static org.apache.flume.sink.elasticsearch.ElasticSearchEventSerializer.charset;
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+import static org.junit.Assert.assertEquals;
+
+public class TestElasticSearchDynamicSerializer {
+
+ @Test
+ public void testRoundTrip() throws Exception {
+ ElasticSearchDynamicSerializer fixture = new ElasticSearchDynamicSerializer();
+ Context context = new Context();
+ fixture.configure(context);
+
+ String message = "test body";
+ Map<String, String> headers = Maps.newHashMap();
+ headers.put("headerNameOne", "headerValueOne");
+ headers.put("headerNameTwo", "headerValueTwo");
+ headers.put("headerNameThree", "headerValueThree");
+ Event event = EventBuilder.withBody(message.getBytes(charset));
+ event.setHeaders(headers);
+
+ XContentBuilder expected = jsonBuilder().startObject();
+ expected.field("body", new String(message.getBytes(), charset));
+ for (String headerName : headers.keySet()) {
+ expected.field(headerName, new String(headers.get(headerName).getBytes(),
+ charset));
+ }
+ expected.endObject();
+
+ XContentBuilder actual = fixture.getContentBuilder(event);
+
+ assertEquals(new String(expected.bytes().array()), new String(actual
+ .bytes().array()));
+
+ }
+}
diff --git a/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/TestElasticSearchIndexRequestBuilderFactory.java b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/TestElasticSearchIndexRequestBuilderFactory.java
new file mode 100644
index 0000000..b62254e
--- /dev/null
+++ b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/TestElasticSearchIndexRequestBuilderFactory.java
@@ -0,0 +1,215 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch;
+
+import com.google.common.collect.Maps;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.conf.ComponentConfiguration;
+import org.apache.flume.conf.sink.SinkConfiguration;
+import org.apache.flume.event.SimpleEvent;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.common.io.BytesStream;
+import org.elasticsearch.common.io.FastByteArrayOutputStream;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Map;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+public class TestElasticSearchIndexRequestBuilderFactory
+ extends AbstractElasticSearchSinkTest {
+
+ private static final Client FAKE_CLIENT = null;
+
+ private EventSerializerIndexRequestBuilderFactory factory;
+
+ private FakeEventSerializer serializer;
+
+ @Before
+ public void setupFactory() throws Exception {
+ serializer = new FakeEventSerializer();
+ factory = new EventSerializerIndexRequestBuilderFactory(serializer) {
+ @Override
+ IndexRequestBuilder prepareIndex(Client client) {
+ return new IndexRequestBuilder(FAKE_CLIENT);
+ }
+ };
+ }
+
+ @Test
+ public void shouldUseUtcAsBasisForDateFormat() {
+ assertEquals("Coordinated Universal Time",
+ factory.fastDateFormat.getTimeZone().getDisplayName());
+ }
+
+ @Test
+ public void indexNameShouldBePrefixDashFormattedTimestamp() {
+ long millis = 987654321L;
+ assertEquals("prefix-" + factory.fastDateFormat.format(millis),
+ factory.getIndexName("prefix", millis));
+ }
+
+ @Test
+ public void shouldEnsureTimestampHeaderPresentInTimestampedEvent() {
+ SimpleEvent base = new SimpleEvent();
+
+ TimestampedEvent timestampedEvent = new TimestampedEvent(base);
+ assertEquals(FIXED_TIME_MILLIS, timestampedEvent.getTimestamp());
+ assertEquals(String.valueOf(FIXED_TIME_MILLIS),
+ timestampedEvent.getHeaders().get("timestamp"));
+ }
+
+ @Test
+ public void shouldUseExistingTimestampHeaderInTimestampedEvent() {
+ SimpleEvent base = new SimpleEvent();
+ Map<String, String> headersWithTimestamp = Maps.newHashMap();
+ headersWithTimestamp.put("timestamp", "-321");
+ base.setHeaders(headersWithTimestamp );
+
+ TimestampedEvent timestampedEvent = new TimestampedEvent(base);
+ assertEquals(-321L, timestampedEvent.getTimestamp());
+ assertEquals("-321", timestampedEvent.getHeaders().get("timestamp"));
+ }
+
+ @Test
+ public void shouldUseExistingAtTimestampHeaderInTimestampedEvent() {
+ SimpleEvent base = new SimpleEvent();
+ Map<String, String> headersWithTimestamp = Maps.newHashMap();
+ headersWithTimestamp.put("@timestamp", "-999");
+ base.setHeaders(headersWithTimestamp );
+
+ TimestampedEvent timestampedEvent = new TimestampedEvent(base);
+ assertEquals(-999L, timestampedEvent.getTimestamp());
+ assertEquals("-999", timestampedEvent.getHeaders().get("@timestamp"));
+ assertNull(timestampedEvent.getHeaders().get("timestamp"));
+ }
+
+ @Test
+ public void shouldPreserveBodyAndNonTimestampHeadersInTimestampedEvent() {
+ SimpleEvent base = new SimpleEvent();
+ base.setBody(new byte[] {1,2,3,4});
+ Map<String, String> headersWithTimestamp = Maps.newHashMap();
+ headersWithTimestamp.put("foo", "bar");
+ base.setHeaders(headersWithTimestamp );
+
+ TimestampedEvent timestampedEvent = new TimestampedEvent(base);
+ assertEquals("bar", timestampedEvent.getHeaders().get("foo"));
+ assertArrayEquals(base.getBody(), timestampedEvent.getBody());
+ }
+
+ @Test
+ public void shouldSetIndexNameTypeAndSerializedEventIntoIndexRequest()
+ throws Exception {
+
+ String indexPrefix = "qwerty";
+ String indexType = "uiop";
+ Event event = new SimpleEvent();
+
+ IndexRequestBuilder indexRequestBuilder = factory.createIndexRequest(
+ FAKE_CLIENT, indexPrefix, indexType, event);
+
+ assertEquals(indexPrefix + '-'
+ + ElasticSearchIndexRequestBuilderFactory.df.format(FIXED_TIME_MILLIS),
+ indexRequestBuilder.request().index());
+ assertEquals(indexType, indexRequestBuilder.request().type());
+ assertArrayEquals(FakeEventSerializer.FAKE_BYTES,
+ indexRequestBuilder.request().source().array());
+ }
+
+ @Test
+ public void shouldSetIndexNameFromTimestampHeaderWhenPresent()
+ throws Exception {
+ String indexPrefix = "qwerty";
+ String indexType = "uiop";
+ Event event = new SimpleEvent();
+ event.getHeaders().put("timestamp", "1213141516");
+
+ IndexRequestBuilder indexRequestBuilder = factory.createIndexRequest(
+ null, indexPrefix, indexType, event);
+
+ assertEquals(indexPrefix + '-'
+ + ElasticSearchIndexRequestBuilderFactory.df.format(1213141516L),
+ indexRequestBuilder.request().index());
+ }
+
+ @Test
+ public void shouldSetIndexNameTypeFromHeaderWhenPresent()
+ throws Exception {
+ String indexPrefix = "%{index-name}";
+ String indexType = "%{index-type}";
+ String indexValue = "testing-index-name-from-headers";
+ String typeValue = "testing-index-type-from-headers";
+
+ Event event = new SimpleEvent();
+ event.getHeaders().put("index-name", indexValue);
+ event.getHeaders().put("index-type", typeValue);
+
+ IndexRequestBuilder indexRequestBuilder = factory.createIndexRequest(
+ null, indexPrefix, indexType, event);
+
+ assertEquals(indexValue + '-'
+ + ElasticSearchIndexRequestBuilderFactory.df.format(FIXED_TIME_MILLIS),
+ indexRequestBuilder.request().index());
+ assertEquals(typeValue, indexRequestBuilder.request().type());
+ }
+
+ @Test
+ public void shouldConfigureEventSerializer() throws Exception {
+ assertFalse(serializer.configuredWithContext);
+ factory.configure(new Context());
+ assertTrue(serializer.configuredWithContext);
+
+ assertFalse(serializer.configuredWithComponentConfiguration);
+ factory.configure(new SinkConfiguration("name"));
+ assertTrue(serializer.configuredWithComponentConfiguration);
+ }
+
+ static class FakeEventSerializer implements ElasticSearchEventSerializer {
+
+ static final byte[] FAKE_BYTES = new byte[]{9, 8, 7, 6};
+ boolean configuredWithContext;
+ boolean configuredWithComponentConfiguration;
+
+ @Override
+ public BytesStream getContentBuilder(Event event) throws IOException {
+ FastByteArrayOutputStream fbaos = new FastByteArrayOutputStream(4);
+ fbaos.write(FAKE_BYTES);
+ return fbaos;
+ }
+
+ @Override
+ public void configure(Context arg0) {
+ configuredWithContext = true;
+ }
+
+ @Override
+ public void configure(ComponentConfiguration arg0) {
+ configuredWithComponentConfiguration = true;
+ }
+ }
+
+}
diff --git a/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/TestElasticSearchLogStashEventSerializer.java b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/TestElasticSearchLogStashEventSerializer.java
new file mode 100644
index 0000000..65b4dab
--- /dev/null
+++ b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/TestElasticSearchLogStashEventSerializer.java
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch;
+
+import com.google.gson.JsonParser;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.event.EventBuilder;
+import org.elasticsearch.common.collect.Maps;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.junit.Test;
+
+import java.util.Date;
+import java.util.Map;
+
+import static org.apache.flume.sink.elasticsearch.ElasticSearchEventSerializer.charset;
+import static org.elasticsearch.common.xcontent.XContentFactory.jsonBuilder;
+import static org.junit.Assert.assertEquals;
+
+public class TestElasticSearchLogStashEventSerializer {
+
+ @Test
+ public void testRoundTrip() throws Exception {
+ ElasticSearchLogStashEventSerializer fixture = new ElasticSearchLogStashEventSerializer();
+ Context context = new Context();
+ fixture.configure(context);
+
+ String message = "test body";
+ Map<String, String> headers = Maps.newHashMap();
+ long timestamp = System.currentTimeMillis();
+ headers.put("timestamp", String.valueOf(timestamp));
+ headers.put("source", "flume_tail_src");
+ headers.put("host", "test@localhost");
+ headers.put("src_path", "/tmp/test");
+ headers.put("headerNameOne", "headerValueOne");
+ headers.put("headerNameTwo", "headerValueTwo");
+ headers.put("type", "sometype");
+ Event event = EventBuilder.withBody(message.getBytes(charset));
+ event.setHeaders(headers);
+
+ XContentBuilder expected = jsonBuilder().startObject();
+ expected.field("@message", new String(message.getBytes(), charset));
+ expected.field("@timestamp", new Date(timestamp));
+ expected.field("@source", "flume_tail_src");
+ expected.field("@type", "sometype");
+ expected.field("@source_host", "test@localhost");
+ expected.field("@source_path", "/tmp/test");
+
+ expected.startObject("@fields");
+ expected.field("timestamp", String.valueOf(timestamp));
+ expected.field("src_path", "/tmp/test");
+ expected.field("host", "test@localhost");
+ expected.field("headerNameTwo", "headerValueTwo");
+ expected.field("source", "flume_tail_src");
+ expected.field("headerNameOne", "headerValueOne");
+ expected.field("type", "sometype");
+ expected.endObject();
+
+ expected.endObject();
+
+ XContentBuilder actual = fixture.getContentBuilder(event);
+
+ JsonParser parser = new JsonParser();
+ assertEquals(parser.parse(expected.string()),parser.parse(actual.string()));
+ }
+
+ @Test
+ public void shouldHandleInvalidJSONDuringComplexParsing() throws Exception {
+ ElasticSearchLogStashEventSerializer fixture = new ElasticSearchLogStashEventSerializer();
+ Context context = new Context();
+ fixture.configure(context);
+
+ String message = "{flume: somethingnotvalid}";
+ Map<String, String> headers = Maps.newHashMap();
+ long timestamp = System.currentTimeMillis();
+ headers.put("timestamp", String.valueOf(timestamp));
+ headers.put("source", "flume_tail_src");
+ headers.put("host", "test@localhost");
+ headers.put("src_path", "/tmp/test");
+ headers.put("headerNameOne", "headerValueOne");
+ headers.put("headerNameTwo", "headerValueTwo");
+ headers.put("type", "sometype");
+ Event event = EventBuilder.withBody(message.getBytes(charset));
+ event.setHeaders(headers);
+
+ XContentBuilder expected = jsonBuilder().startObject();
+ expected.field("@message", new String(message.getBytes(), charset));
+ expected.field("@timestamp", new Date(timestamp));
+ expected.field("@source", "flume_tail_src");
+ expected.field("@type", "sometype");
+ expected.field("@source_host", "test@localhost");
+ expected.field("@source_path", "/tmp/test");
+
+ expected.startObject("@fields");
+ expected.field("timestamp", String.valueOf(timestamp));
+ expected.field("src_path", "/tmp/test");
+ expected.field("host", "test@localhost");
+ expected.field("headerNameTwo", "headerValueTwo");
+ expected.field("source", "flume_tail_src");
+ expected.field("headerNameOne", "headerValueOne");
+ expected.field("type", "sometype");
+ expected.endObject();
+
+ expected.endObject();
+
+ XContentBuilder actual = fixture.getContentBuilder(event);
+
+ JsonParser parser = new JsonParser();
+ assertEquals(parser.parse(expected.string()),parser.parse(actual.string()));
+ }
+}
diff --git a/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/TestElasticSearchSink.java b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/TestElasticSearchSink.java
new file mode 100644
index 0000000..69acc06
--- /dev/null
+++ b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/TestElasticSearchSink.java
@@ -0,0 +1,505 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch;
+
+import org.apache.commons.lang.time.FastDateFormat;
+import org.apache.flume.Channel;
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.Sink.Status;
+import org.apache.flume.Transaction;
+import org.apache.flume.conf.ComponentConfiguration;
+import org.apache.flume.conf.Configurable;
+import org.apache.flume.conf.Configurables;
+import org.apache.flume.event.EventBuilder;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.client.Requests;
+import org.elasticsearch.common.UUID;
+import org.elasticsearch.common.io.BytesStream;
+import org.elasticsearch.common.io.FastByteArrayOutputStream;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.TimeZone;
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.BATCH_SIZE;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.CLUSTER_NAME;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.HOSTNAMES;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.INDEX_NAME;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.INDEX_TYPE;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.SERIALIZER;
+import static org.apache.flume.sink.elasticsearch.ElasticSearchSinkConstants.TTL;
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+public class TestElasticSearchSink extends AbstractElasticSearchSinkTest {
+
+ private ElasticSearchSink fixture;
+
+ @Before
+ public void init() throws Exception {
+ initDefaults();
+ createNodes();
+ fixture = new ElasticSearchSink(true);
+ fixture.setName("ElasticSearchSink-" + UUID.randomUUID().toString());
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ shutdownNodes();
+ }
+
+ @Test
+ public void shouldIndexOneEvent() throws Exception {
+ Configurables.configure(fixture, new Context(parameters));
+ Channel channel = bindAndStartChannel(fixture);
+
+ Transaction tx = channel.getTransaction();
+ tx.begin();
+ Event event = EventBuilder.withBody("event #1 or 1".getBytes());
+ channel.put(event);
+ tx.commit();
+ tx.close();
+
+ fixture.process();
+ fixture.stop();
+ client.admin().indices()
+ .refresh(Requests.refreshRequest(timestampedIndexName)).actionGet();
+
+ assertMatchAllQuery(1, event);
+ assertBodyQuery(1, event);
+ }
+
+ @Test
+ public void shouldIndexInvalidComplexJsonBody() throws Exception {
+ parameters.put(BATCH_SIZE, "3");
+ Configurables.configure(fixture, new Context(parameters));
+ Channel channel = bindAndStartChannel(fixture);
+
+ Transaction tx = channel.getTransaction();
+ tx.begin();
+ Event event1 = EventBuilder.withBody("TEST1 {test}".getBytes());
+ channel.put(event1);
+ Event event2 = EventBuilder.withBody("{test: TEST2 }".getBytes());
+ channel.put(event2);
+ Event event3 = EventBuilder.withBody("{\"test\":{ TEST3 {test} }}".getBytes());
+ channel.put(event3);
+ tx.commit();
+ tx.close();
+
+ fixture.process();
+ fixture.stop();
+ client.admin().indices()
+ .refresh(Requests.refreshRequest(timestampedIndexName)).actionGet();
+
+ assertMatchAllQuery(3);
+ assertSearch(1,
+ performSearch(QueryBuilders.fieldQuery("@message", "TEST1")),
+ null, event1);
+ assertSearch(1,
+ performSearch(QueryBuilders.fieldQuery("@message", "TEST2")),
+ null, event2);
+ assertSearch(1,
+ performSearch(QueryBuilders.fieldQuery("@message", "TEST3")),
+ null, event3);
+ }
+
+ @Test
+ public void shouldIndexComplexJsonEvent() throws Exception {
+ Configurables.configure(fixture, new Context(parameters));
+ Channel channel = bindAndStartChannel(fixture);
+
+ Transaction tx = channel.getTransaction();
+ tx.begin();
+ Event event = EventBuilder.withBody(
+ "{\"event\":\"json content\",\"num\":1}".getBytes());
+ channel.put(event);
+ tx.commit();
+ tx.close();
+
+ fixture.process();
+ fixture.stop();
+ client.admin().indices()
+ .refresh(Requests.refreshRequest(timestampedIndexName)).actionGet();
+
+ Map<String, Object> expectedBody = new HashMap<String, Object>();
+ expectedBody.put("event", "json content");
+ expectedBody.put("num", 1);
+
+ assertSearch(1,
+ performSearch(QueryBuilders.matchAllQuery()), expectedBody, event);
+ assertSearch(1,
+ performSearch(QueryBuilders.fieldQuery("@message.event", "json")),
+ expectedBody, event);
+ }
+
+ @Test
+ public void shouldIndexFiveEvents() throws Exception {
+ // Make it so we only need to call process once
+ parameters.put(BATCH_SIZE, "5");
+ Configurables.configure(fixture, new Context(parameters));
+ Channel channel = bindAndStartChannel(fixture);
+
+ int numberOfEvents = 5;
+ Event[] events = new Event[numberOfEvents];
+
+ Transaction tx = channel.getTransaction();
+ tx.begin();
+ for (int i = 0; i < numberOfEvents; i++) {
+ String body = "event #" + i + " of " + numberOfEvents;
+ Event event = EventBuilder.withBody(body.getBytes());
+ events[i] = event;
+ channel.put(event);
+ }
+ tx.commit();
+ tx.close();
+
+ fixture.process();
+ fixture.stop();
+ client.admin().indices()
+ .refresh(Requests.refreshRequest(timestampedIndexName)).actionGet();
+
+ assertMatchAllQuery(numberOfEvents, events);
+ assertBodyQuery(5, events);
+ }
+
+ @Test
+ public void shouldIndexFiveEventsOverThreeBatches() throws Exception {
+ parameters.put(BATCH_SIZE, "2");
+ Configurables.configure(fixture, new Context(parameters));
+ Channel channel = bindAndStartChannel(fixture);
+
+ int numberOfEvents = 5;
+ Event[] events = new Event[numberOfEvents];
+
+ Transaction tx = channel.getTransaction();
+ tx.begin();
+ for (int i = 0; i < numberOfEvents; i++) {
+ String body = "event #" + i + " of " + numberOfEvents;
+ Event event = EventBuilder.withBody(body.getBytes());
+ events[i] = event;
+ channel.put(event);
+ }
+ tx.commit();
+ tx.close();
+
+ int count = 0;
+ Status status = Status.READY;
+ while (status != Status.BACKOFF) {
+ count++;
+ status = fixture.process();
+ }
+ fixture.stop();
+
+ assertEquals(3, count);
+
+ client.admin().indices()
+ .refresh(Requests.refreshRequest(timestampedIndexName)).actionGet();
+ assertMatchAllQuery(numberOfEvents, events);
+ assertBodyQuery(5, events);
+ }
+
+ @Test
+ public void shouldParseConfiguration() {
+ parameters.put(HOSTNAMES, "10.5.5.27");
+ parameters.put(CLUSTER_NAME, "testing-cluster-name");
+ parameters.put(INDEX_NAME, "testing-index-name");
+ parameters.put(INDEX_TYPE, "testing-index-type");
+ parameters.put(TTL, "10");
+
+ fixture = new ElasticSearchSink();
+ fixture.configure(new Context(parameters));
+
+ String[] expected = { "10.5.5.27" };
+
+ assertEquals("testing-cluster-name", fixture.getClusterName());
+ assertEquals("testing-index-name", fixture.getIndexName());
+ assertEquals("testing-index-type", fixture.getIndexType());
+ assertEquals(TimeUnit.DAYS.toMillis(10), fixture.getTTLMs());
+ assertArrayEquals(expected, fixture.getServerAddresses());
+ }
+
+ @Test
+ public void shouldParseConfigurationUsingDefaults() {
+ parameters.put(HOSTNAMES, "10.5.5.27");
+ parameters.remove(INDEX_NAME);
+ parameters.remove(INDEX_TYPE);
+ parameters.remove(CLUSTER_NAME);
+
+ fixture = new ElasticSearchSink();
+ fixture.configure(new Context(parameters));
+
+ String[] expected = { "10.5.5.27" };
+
+ assertEquals(DEFAULT_INDEX_NAME, fixture.getIndexName());
+ assertEquals(DEFAULT_INDEX_TYPE, fixture.getIndexType());
+ assertEquals(DEFAULT_CLUSTER_NAME, fixture.getClusterName());
+ assertArrayEquals(expected, fixture.getServerAddresses());
+ }
+
+ @Test
+ public void shouldParseMultipleHostUsingDefaultPorts() {
+ parameters.put(HOSTNAMES, "10.5.5.27,10.5.5.28,10.5.5.29");
+
+ fixture = new ElasticSearchSink();
+ fixture.configure(new Context(parameters));
+
+ String[] expected = { "10.5.5.27", "10.5.5.28", "10.5.5.29" };
+
+ assertArrayEquals(expected, fixture.getServerAddresses());
+ }
+
+ @Test
+ public void shouldParseMultipleHostWithWhitespacesUsingDefaultPorts() {
+ parameters.put(HOSTNAMES, " 10.5.5.27 , 10.5.5.28 , 10.5.5.29 ");
+
+ fixture = new ElasticSearchSink();
+ fixture.configure(new Context(parameters));
+
+ String[] expected = { "10.5.5.27", "10.5.5.28", "10.5.5.29" };
+
+ assertArrayEquals(expected, fixture.getServerAddresses());
+ }
+
+ @Test
+ public void shouldParseMultipleHostAndPorts() {
+ parameters.put(HOSTNAMES, "10.5.5.27:9300,10.5.5.28:9301,10.5.5.29:9302");
+
+ fixture = new ElasticSearchSink();
+ fixture.configure(new Context(parameters));
+
+ String[] expected = { "10.5.5.27:9300", "10.5.5.28:9301", "10.5.5.29:9302" };
+
+ assertArrayEquals(expected, fixture.getServerAddresses());
+ }
+
+ @Test
+ public void shouldParseMultipleHostAndPortsWithWhitespaces() {
+ parameters.put(HOSTNAMES,
+ " 10.5.5.27 : 9300 , 10.5.5.28 : 9301 , 10.5.5.29 : 9302 ");
+
+ fixture = new ElasticSearchSink();
+ fixture.configure(new Context(parameters));
+
+ String[] expected = { "10.5.5.27:9300", "10.5.5.28:9301", "10.5.5.29:9302" };
+
+ assertArrayEquals(expected, fixture.getServerAddresses());
+ }
+
+ @Test
+ public void shouldAllowCustomElasticSearchIndexRequestBuilderFactory()
+ throws Exception {
+ parameters.put(SERIALIZER,
+ CustomElasticSearchIndexRequestBuilderFactory.class.getName());
+
+ fixture.configure(new Context(parameters));
+
+ Channel channel = bindAndStartChannel(fixture);
+ Transaction tx = channel.getTransaction();
+ tx.begin();
+ String body = "{ foo: \"bar\" }";
+ Event event = EventBuilder.withBody(body.getBytes());
+ channel.put(event);
+ tx.commit();
+ tx.close();
+
+ fixture.process();
+ fixture.stop();
+
+ assertEquals(fixture.getIndexName() + "-05_17_36_789",
+ CustomElasticSearchIndexRequestBuilderFactory.actualIndexName);
+ assertEquals(fixture.getIndexType(),
+ CustomElasticSearchIndexRequestBuilderFactory.actualIndexType);
+ assertArrayEquals(event.getBody(),
+ CustomElasticSearchIndexRequestBuilderFactory.actualEventBody);
+ assertTrue(CustomElasticSearchIndexRequestBuilderFactory.hasContext);
+ }
+
+ @Test
+ public void shouldParseFullyQualifiedTTLs() {
+ Map<String, Long> testTTLMap = new HashMap<String, Long>();
+ testTTLMap.put("1ms", Long.valueOf(1));
+ testTTLMap.put("1s", Long.valueOf(1000));
+ testTTLMap.put("1m", Long.valueOf(60000));
+ testTTLMap.put("1h", Long.valueOf(3600000));
+ testTTLMap.put("1d", Long.valueOf(86400000));
+ testTTLMap.put("1w", Long.valueOf(604800000));
+ testTTLMap.put("1", Long.valueOf(86400000));
+
+ parameters.put(HOSTNAMES, "10.5.5.27");
+ parameters.put(CLUSTER_NAME, "testing-cluster-name");
+ parameters.put(INDEX_NAME, "testing-index-name");
+ parameters.put(INDEX_TYPE, "testing-index-type");
+
+ for (String ttl : testTTLMap.keySet()) {
+ parameters.put(TTL, ttl);
+ fixture = new ElasticSearchSink();
+ fixture.configure(new Context(parameters));
+
+ String[] expected = { "10.5.5.27" };
+ assertEquals("testing-cluster-name", fixture.getClusterName());
+ assertEquals("testing-index-name", fixture.getIndexName());
+ assertEquals("testing-index-type", fixture.getIndexType());
+ assertEquals((long) testTTLMap.get(ttl), fixture.getTTLMs());
+ assertArrayEquals(expected, fixture.getServerAddresses());
+
+ }
+ }
+
+ public static final class CustomElasticSearchIndexRequestBuilderFactory
+ extends AbstractElasticSearchIndexRequestBuilderFactory {
+
+ static String actualIndexName;
+ static String actualIndexType;
+ static byte[] actualEventBody;
+ static boolean hasContext;
+
+ public CustomElasticSearchIndexRequestBuilderFactory() {
+ super(FastDateFormat.getInstance("HH_mm_ss_SSS", TimeZone.getTimeZone("EST5EDT")));
+ }
+
+ @Override
+ protected void prepareIndexRequest(IndexRequestBuilder indexRequest, String indexName,
+ String indexType, Event event) throws IOException {
+ actualIndexName = indexName;
+ actualIndexType = indexType;
+ actualEventBody = event.getBody();
+ indexRequest.setIndex(indexName).setType(indexType).setSource(event.getBody());
+ }
+
+ @Override
+ public void configure(Context arg0) {
+ hasContext = true;
+ }
+
+ @Override
+ public void configure(ComponentConfiguration arg0) {
+ //no-op
+ }
+ }
+
+ @Test
+ public void shouldFailToConfigureWithInvalidSerializerClass()
+ throws Exception {
+
+ parameters.put(SERIALIZER, "java.lang.String");
+ try {
+ Configurables.configure(fixture, new Context(parameters));
+ } catch (ClassCastException e) {
+ // expected
+ }
+
+ parameters.put(SERIALIZER, FakeConfigurable.class.getName());
+ try {
+ Configurables.configure(fixture, new Context(parameters));
+ } catch (IllegalArgumentException e) {
+ // expected
+ }
+ }
+
+ @Test
+ public void shouldUseSpecifiedSerializer() throws Exception {
+ Context context = new Context();
+ context.put(SERIALIZER,
+ "org.apache.flume.sink.elasticsearch.FakeEventSerializer");
+
+ assertNull(fixture.getEventSerializer());
+ fixture.configure(context);
+ assertTrue(fixture.getEventSerializer() instanceof FakeEventSerializer);
+ }
+
+ @Test
+ public void shouldUseSpecifiedIndexNameBuilder() throws Exception {
+ Context context = new Context();
+ context.put(ElasticSearchSinkConstants.INDEX_NAME_BUILDER,
+ "org.apache.flume.sink.elasticsearch.FakeIndexNameBuilder");
+
+ assertNull(fixture.getIndexNameBuilder());
+ fixture.configure(context);
+ assertTrue(fixture.getIndexNameBuilder() instanceof FakeIndexNameBuilder);
+ }
+
+ public static class FakeConfigurable implements Configurable {
+ @Override
+ public void configure(Context arg0) {
+ // no-op
+ }
+ }
+}
+
+/**
+ * Internal class. Fake event serializer used for tests
+ */
+class FakeEventSerializer implements ElasticSearchEventSerializer {
+
+ static final byte[] FAKE_BYTES = new byte[] { 9, 8, 7, 6 };
+ boolean configuredWithContext;
+ boolean configuredWithComponentConfiguration;
+
+ @Override
+ public BytesStream getContentBuilder(Event event) throws IOException {
+ FastByteArrayOutputStream fbaos = new FastByteArrayOutputStream(4);
+ fbaos.write(FAKE_BYTES);
+ return fbaos;
+ }
+
+ @Override
+ public void configure(Context arg0) {
+ configuredWithContext = true;
+ }
+
+ @Override
+ public void configure(ComponentConfiguration arg0) {
+ configuredWithComponentConfiguration = true;
+ }
+}
+
+/**
+ * Internal class. Fake index name builder used only for tests.
+ */
+class FakeIndexNameBuilder implements IndexNameBuilder {
+
+ static final String INDEX_NAME = "index_name";
+
+ @Override
+ public String getIndexName(Event event) {
+ return INDEX_NAME;
+ }
+
+ @Override
+ public String getIndexPrefix(Event event) {
+ return INDEX_NAME;
+ }
+
+ @Override
+ public void configure(Context context) {
+ }
+
+ @Override
+ public void configure(ComponentConfiguration conf) {
+ }
+}
diff --git a/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/TestElasticSearchSinkCreation.java b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/TestElasticSearchSinkCreation.java
new file mode 100644
index 0000000..2a36439
--- /dev/null
+++ b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/TestElasticSearchSinkCreation.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch;
+
+import org.apache.flume.FlumeException;
+import org.apache.flume.Sink;
+import org.apache.flume.SinkFactory;
+import org.apache.flume.sink.DefaultSinkFactory;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+public class TestElasticSearchSinkCreation {
+
+ private SinkFactory sinkFactory;
+
+ @Before
+ public void setUp() {
+ sinkFactory = new DefaultSinkFactory();
+ }
+
+ private void verifySinkCreation(String name, String type,
+ Class<?> typeClass) throws FlumeException {
+ Sink sink = sinkFactory.create(name, type);
+ Assert.assertNotNull(sink);
+ Assert.assertTrue(typeClass.isInstance(sink));
+ }
+
+ @Test
+ public void testSinkCreation() {
+ verifySinkCreation("elasticsearch-sink", "elasticsearch", ElasticSearchSink.class);
+ }
+}
diff --git a/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/TimeBasedIndexNameBuilderTest.java b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/TimeBasedIndexNameBuilderTest.java
new file mode 100644
index 0000000..678342a
--- /dev/null
+++ b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/TimeBasedIndexNameBuilderTest.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch;
+
+import org.apache.flume.Context;
+import org.apache.flume.Event;
+import org.apache.flume.event.SimpleEvent;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+
+public class TimeBasedIndexNameBuilderTest {
+
+ private TimeBasedIndexNameBuilder indexNameBuilder;
+
+ @Before
+ public void setUp() throws Exception {
+ Context context = new Context();
+ context.put(ElasticSearchSinkConstants.INDEX_NAME, "prefix");
+ indexNameBuilder = new TimeBasedIndexNameBuilder();
+ indexNameBuilder.configure(context);
+ }
+
+ @Test
+ public void shouldUseUtcAsBasisForDateFormat() {
+ assertEquals("Coordinated Universal Time",
+ indexNameBuilder.getFastDateFormat().getTimeZone().getDisplayName());
+ }
+
+ @Test
+ public void indexNameShouldBePrefixDashFormattedTimestamp() {
+ long time = 987654321L;
+ Event event = new SimpleEvent();
+ Map<String, String> headers = new HashMap<String, String>();
+ headers.put("timestamp", Long.toString(time));
+ event.setHeaders(headers);
+ assertEquals("prefix-" + indexNameBuilder.getFastDateFormat().format(time),
+ indexNameBuilder.getIndexName(event));
+ }
+}
diff --git a/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/TimestampedEventTest.java b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/TimestampedEventTest.java
new file mode 100644
index 0000000..bef2ac6
--- /dev/null
+++ b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/TimestampedEventTest.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch;
+
+import com.google.common.collect.Maps;
+import org.apache.flume.event.SimpleEvent;
+import org.joda.time.DateTimeUtils;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Map;
+
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+
+public class TimestampedEventTest {
+ static final long FIXED_TIME_MILLIS = 123456789L;
+
+ @Before
+ public void setFixedJodaTime() {
+ DateTimeUtils.setCurrentMillisFixed(FIXED_TIME_MILLIS);
+ }
+
+ @Test
+ public void shouldEnsureTimestampHeaderPresentInTimestampedEvent() {
+ SimpleEvent base = new SimpleEvent();
+
+ TimestampedEvent timestampedEvent = new TimestampedEvent(base);
+ assertEquals(FIXED_TIME_MILLIS, timestampedEvent.getTimestamp());
+ assertEquals(String.valueOf(FIXED_TIME_MILLIS),
+ timestampedEvent.getHeaders().get("timestamp"));
+ }
+
+ @Test
+ public void shouldUseExistingTimestampHeaderInTimestampedEvent() {
+ SimpleEvent base = new SimpleEvent();
+ Map<String, String> headersWithTimestamp = Maps.newHashMap();
+ headersWithTimestamp.put("timestamp", "-321");
+ base.setHeaders(headersWithTimestamp );
+
+ TimestampedEvent timestampedEvent = new TimestampedEvent(base);
+ assertEquals(-321L, timestampedEvent.getTimestamp());
+ assertEquals("-321", timestampedEvent.getHeaders().get("timestamp"));
+ }
+
+ @Test
+ public void shouldUseExistingAtTimestampHeaderInTimestampedEvent() {
+ SimpleEvent base = new SimpleEvent();
+ Map<String, String> headersWithTimestamp = Maps.newHashMap();
+ headersWithTimestamp.put("@timestamp", "-999");
+ base.setHeaders(headersWithTimestamp );
+
+ TimestampedEvent timestampedEvent = new TimestampedEvent(base);
+ assertEquals(-999L, timestampedEvent.getTimestamp());
+ assertEquals("-999", timestampedEvent.getHeaders().get("@timestamp"));
+ assertNull(timestampedEvent.getHeaders().get("timestamp"));
+ }
+
+ @Test
+ public void shouldPreserveBodyAndNonTimestampHeadersInTimestampedEvent() {
+ SimpleEvent base = new SimpleEvent();
+ base.setBody(new byte[] {1,2,3,4});
+ Map<String, String> headersWithTimestamp = Maps.newHashMap();
+ headersWithTimestamp.put("foo", "bar");
+ base.setHeaders(headersWithTimestamp );
+
+ TimestampedEvent timestampedEvent = new TimestampedEvent(base);
+ assertEquals("bar", timestampedEvent.getHeaders().get("foo"));
+ assertArrayEquals(base.getBody(), timestampedEvent.getBody());
+ }
+}
diff --git a/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/client/RoundRobinListTest.java b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/client/RoundRobinListTest.java
new file mode 100644
index 0000000..0d1d092
--- /dev/null
+++ b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/client/RoundRobinListTest.java
@@ -0,0 +1,42 @@
+/*
+ * Copyright 2014 Apache Software Foundation.
+ *
+ * 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 org.apache.flume.sink.elasticsearch.client;
+
+import java.util.Arrays;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class RoundRobinListTest {
+
+ private RoundRobinList<String> fixture;
+
+ @Before
+ public void setUp() {
+ fixture = new RoundRobinList<String>(Arrays.asList("test1", "test2"));
+ }
+
+ @Test
+ public void shouldReturnNextElement() {
+ assertEquals("test1", fixture.get());
+ assertEquals("test2", fixture.get());
+ assertEquals("test1", fixture.get());
+ assertEquals("test2", fixture.get());
+ assertEquals("test1", fixture.get());
+ }
+}
diff --git a/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/client/TestElasticSearchClientFactory.java b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/client/TestElasticSearchClientFactory.java
new file mode 100644
index 0000000..c3f07b0
--- /dev/null
+++ b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/client/TestElasticSearchClientFactory.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch.client;
+
+import org.apache.flume.sink.elasticsearch.ElasticSearchEventSerializer;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+
+import static org.hamcrest.core.IsInstanceOf.instanceOf;
+import static org.junit.Assert.assertThat;
+import static org.mockito.MockitoAnnotations.initMocks;
+
+public class TestElasticSearchClientFactory {
+
+ ElasticSearchClientFactory factory;
+
+ @Mock
+ ElasticSearchEventSerializer serializer;
+
+ @Before
+ public void setUp() {
+ initMocks(this);
+ factory = new ElasticSearchClientFactory();
+ }
+
+ @Test
+ public void shouldReturnTransportClient() throws Exception {
+ String[] hostNames = { "127.0.0.1" };
+ Object o = factory.getClient(ElasticSearchClientFactory.TransportClient,
+ hostNames, "test", serializer, null);
+ assertThat(o, instanceOf(ElasticSearchTransportClient.class));
+ }
+
+ @Test
+ public void shouldReturnRestClient() throws NoSuchClientTypeException {
+ String[] hostNames = { "127.0.0.1" };
+ Object o = factory.getClient(ElasticSearchClientFactory.RestClient,
+ hostNames, "test", serializer, null);
+ assertThat(o, instanceOf(ElasticSearchRestClient.class));
+ }
+
+ @Test(expected = NoSuchClientTypeException.class)
+ public void shouldThrowNoSuchClientTypeException() throws NoSuchClientTypeException {
+ String[] hostNames = { "127.0.0.1" };
+ factory.getClient("not_existing_client", hostNames, "test", null, null);
+ }
+}
diff --git a/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/client/TestElasticSearchRestClient.java b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/client/TestElasticSearchRestClient.java
new file mode 100644
index 0000000..9551c81
--- /dev/null
+++ b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/client/TestElasticSearchRestClient.java
@@ -0,0 +1,180 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch.client;
+
+import com.google.common.base.Splitter;
+import com.google.gson.JsonObject;
+import com.google.gson.JsonParser;
+import org.apache.flume.Event;
+import org.apache.flume.EventDeliveryException;
+import org.apache.flume.sink.elasticsearch.ElasticSearchEventSerializer;
+import org.apache.flume.sink.elasticsearch.IndexNameBuilder;
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpResponse;
+import org.apache.http.HttpStatus;
+import org.apache.http.StatusLine;
+import org.apache.http.client.HttpClient;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.util.EntityUtils;
+import org.elasticsearch.common.bytes.BytesArray;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.io.BytesStream;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.List;
+
+import static junit.framework.Assert.assertEquals;
+import static junit.framework.Assert.assertTrue;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.isA;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+import static org.mockito.MockitoAnnotations.initMocks;
+
+public class TestElasticSearchRestClient {
+
+ private ElasticSearchRestClient fixture;
+
+ @Mock
+ private ElasticSearchEventSerializer serializer;
+
+ @Mock
+ private IndexNameBuilder nameBuilder;
+
+ @Mock
+ private Event event;
+
+ @Mock
+ private HttpClient httpClient;
+
+ @Mock
+ private HttpResponse httpResponse;
+
+ @Mock
+ private StatusLine httpStatus;
+
+ @Mock
+ private HttpEntity httpEntity;
+
+ private static final String INDEX_NAME = "foo_index";
+ private static final String MESSAGE_CONTENT = "{\"body\":\"test\"}";
+ private static final String[] HOSTS = {"host1", "host2"};
+
+ @Before
+ public void setUp() throws IOException {
+ initMocks(this);
+ BytesReference bytesReference = mock(BytesReference.class);
+ BytesStream bytesStream = mock(BytesStream.class);
+
+ when(nameBuilder.getIndexName(any(Event.class))).thenReturn(INDEX_NAME);
+ when(bytesReference.toBytesArray()).thenReturn(new BytesArray(MESSAGE_CONTENT));
+ when(bytesStream.bytes()).thenReturn(bytesReference);
+ when(serializer.getContentBuilder(any(Event.class))).thenReturn(bytesStream);
+ fixture = new ElasticSearchRestClient(HOSTS, serializer, httpClient);
+ }
+
+ @Test
+ public void shouldAddNewEventWithoutTTL() throws Exception {
+ ArgumentCaptor<HttpPost> argument = ArgumentCaptor.forClass(HttpPost.class);
+
+ when(httpStatus.getStatusCode()).thenReturn(HttpStatus.SC_OK);
+ when(httpResponse.getStatusLine()).thenReturn(httpStatus);
+ when(httpClient.execute(any(HttpUriRequest.class))).thenReturn(httpResponse);
+
+ fixture.addEvent(event, nameBuilder, "bar_type", -1);
+ fixture.execute();
+
+ verify(httpClient).execute(isA(HttpUriRequest.class));
+ verify(httpClient).execute(argument.capture());
+
+ assertEquals("http://host1/_bulk", argument.getValue().getURI().toString());
+ assertTrue(verifyJsonEvents("{\"index\":{\"_type\":\"bar_type\", \"_index\":\"foo_index\"}}\n",
+ MESSAGE_CONTENT, EntityUtils.toString(argument.getValue().getEntity())));
+ }
+
+ @Test
+ public void shouldAddNewEventWithTTL() throws Exception {
+ ArgumentCaptor<HttpPost> argument = ArgumentCaptor.forClass(HttpPost.class);
+
+ when(httpStatus.getStatusCode()).thenReturn(HttpStatus.SC_OK);
+ when(httpResponse.getStatusLine()).thenReturn(httpStatus);
+ when(httpClient.execute(any(HttpUriRequest.class))).thenReturn(httpResponse);
+
+ fixture.addEvent(event, nameBuilder, "bar_type", 123);
+ fixture.execute();
+
+ verify(httpClient).execute(isA(HttpUriRequest.class));
+ verify(httpClient).execute(argument.capture());
+
+ assertEquals("http://host1/_bulk", argument.getValue().getURI().toString());
+ assertTrue(verifyJsonEvents(
+ "{\"index\":{\"_type\":\"bar_type\",\"_index\":\"foo_index\",\"_ttl\":\"123\"}}\n",
+ MESSAGE_CONTENT, EntityUtils.toString(argument.getValue().getEntity())));
+ }
+
+ private boolean verifyJsonEvents(String expectedIndex, String expectedBody, String actual) {
+ Iterator<String> it = Splitter.on("\n").split(actual).iterator();
+ JsonParser parser = new JsonParser();
+ JsonObject[] arr = new JsonObject[2];
+ for (int i = 0; i < 2; i++) {
+ arr[i] = (JsonObject) parser.parse(it.next());
+ }
+ return arr[0].equals(parser.parse(expectedIndex)) && arr[1].equals(parser.parse(expectedBody));
+ }
+
+ @Test(expected = EventDeliveryException.class)
+ public void shouldThrowEventDeliveryException() throws Exception {
+ ArgumentCaptor<HttpPost> argument = ArgumentCaptor.forClass(HttpPost.class);
+
+ when(httpStatus.getStatusCode()).thenReturn(HttpStatus.SC_INTERNAL_SERVER_ERROR);
+ when(httpResponse.getStatusLine()).thenReturn(httpStatus);
+ when(httpClient.execute(any(HttpUriRequest.class))).thenReturn(httpResponse);
+
+ fixture.addEvent(event, nameBuilder, "bar_type", 123);
+ fixture.execute();
+ }
+
+ @Test()
+ public void shouldRetryBulkOperation() throws Exception {
+ ArgumentCaptor<HttpPost> argument = ArgumentCaptor.forClass(HttpPost.class);
+
+ when(httpStatus.getStatusCode()).thenReturn(HttpStatus.SC_INTERNAL_SERVER_ERROR,
+ HttpStatus.SC_OK);
+ when(httpResponse.getStatusLine()).thenReturn(httpStatus);
+ when(httpClient.execute(any(HttpUriRequest.class))).thenReturn(httpResponse);
+
+ fixture.addEvent(event, nameBuilder, "bar_type", 123);
+ fixture.execute();
+
+ verify(httpClient, times(2)).execute(isA(HttpUriRequest.class));
+ verify(httpClient, times(2)).execute(argument.capture());
+
+ List<HttpPost> allValues = argument.getAllValues();
+ assertEquals("http://host1/_bulk", allValues.get(0).getURI().toString());
+ assertEquals("http://host2/_bulk", allValues.get(1).getURI().toString());
+ }
+}
diff --git a/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/client/TestElasticSearchTransportClient.java b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/client/TestElasticSearchTransportClient.java
new file mode 100644
index 0000000..b7b8e74
--- /dev/null
+++ b/flume-elasticsearch-sink/src/test/java/org/apache/flume/sink/elasticsearch/client/TestElasticSearchTransportClient.java
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.flume.sink.elasticsearch.client;
+
+import org.apache.flume.Event;
+import org.apache.flume.EventDeliveryException;
+import org.apache.flume.sink.elasticsearch.ElasticSearchEventSerializer;
+import org.apache.flume.sink.elasticsearch.IndexNameBuilder;
+import org.elasticsearch.action.ListenableActionFuture;
+import org.elasticsearch.action.bulk.BulkRequestBuilder;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.index.IndexRequestBuilder;
+import org.elasticsearch.client.Client;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.io.BytesStream;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mock;
+
+import java.io.IOException;
+
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.*;
+import static org.mockito.MockitoAnnotations.initMocks;
+
+public class TestElasticSearchTransportClient {
+
+ private ElasticSearchTransportClient fixture;
+
+ @Mock
+ private ElasticSearchEventSerializer serializer;
+
+ @Mock
+ private IndexNameBuilder nameBuilder;
+
+ @Mock
+ private Client elasticSearchClient;
+
+ @Mock
+ private BulkRequestBuilder bulkRequestBuilder;
+
+ @Mock
+ private IndexRequestBuilder indexRequestBuilder;
+
+ @Mock
+ private Event event;
+
+ @Before
+ public void setUp() throws IOException {
+ initMocks(this);
+ BytesReference bytesReference = mock(BytesReference.class);
+ BytesStream bytesStream = mock(BytesStream.class);
+
+ when(nameBuilder.getIndexName(any(Event.class))).thenReturn("foo_index");
+ when(bytesReference.toBytes()).thenReturn("{\"body\":\"test\"}".getBytes());
+ when(bytesStream.bytes()).thenReturn(bytesReference);
+ when(serializer.getContentBuilder(any(Event.class)))
+ .thenReturn(bytesStream);
+ when(elasticSearchClient.prepareIndex(anyString(), anyString()))
+ .thenReturn(indexRequestBuilder);
+ when(indexRequestBuilder.setSource(bytesReference)).thenReturn(
+ indexRequestBuilder);
+
+ fixture = new ElasticSearchTransportClient(elasticSearchClient, serializer);
+ fixture.setBulkRequestBuilder(bulkRequestBuilder);
+ }
+
+ @Test
+ public void shouldAddNewEventWithoutTTL() throws Exception {
+ fixture.addEvent(event, nameBuilder, "bar_type", -1);
+ verify(indexRequestBuilder).setSource(
+ serializer.getContentBuilder(event).bytes());
+ verify(bulkRequestBuilder).add(indexRequestBuilder);
+ }
+
+ @Test
+ public void shouldAddNewEventWithTTL() throws Exception {
+ fixture.addEvent(event, nameBuilder, "bar_type", 10);
+ verify(indexRequestBuilder).setTTL(10);
+ verify(indexRequestBuilder).setSource(
+ serializer.getContentBuilder(event).bytes());
+ }
+
+ @Test
+ public void shouldExecuteBulkRequestBuilder() throws Exception {
+ ListenableActionFuture<BulkResponse> action =
+ (ListenableActionFuture<BulkResponse>) mock(ListenableActionFuture.class);
+ BulkResponse response = mock(BulkResponse.class);
+ when(bulkRequestBuilder.execute()).thenReturn(action);
+ when(action.actionGet()).thenReturn(response);
+ when(response.hasFailures()).thenReturn(false);
+
+ fixture.addEvent(event, nameBuilder, "bar_type", 10);
+ fixture.execute();
+ verify(bulkRequestBuilder).execute();
+ }
+
+ @Test(expected = EventDeliveryException.class)
+ public void shouldThrowExceptionOnExecuteFailed() throws Exception {
+ ListenableActionFuture<BulkResponse> action =
+ (ListenableActionFuture<BulkResponse>) mock(ListenableActionFuture.class);
+ BulkResponse response = mock(BulkResponse.class);
+ when(bulkRequestBuilder.execute()).thenReturn(action);
+ when(action.actionGet()).thenReturn(response);
+ when(response.hasFailures()).thenReturn(true);
+
+ fixture.addEvent(event, nameBuilder, "bar_type", 10);
+ fixture.execute();
+ }
+}
diff --git a/flume-elasticsearch-sink/src/test/resources/log4j2.xml b/flume-elasticsearch-sink/src/test/resources/log4j2.xml
new file mode 100644
index 0000000..5676916
--- /dev/null
+++ b/flume-elasticsearch-sink/src/test/resources/log4j2.xml
@@ -0,0 +1,31 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one or more
+ contributor license agreements. See the NOTICE file distributed with
+ this work for additional information regarding copyright ownership.
+ The ASF licenses this file to You 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.
+
+-->
+<Configuration status="OFF">
+ <Appenders>
+ <Console name="Console" target="SYSTEM_OUT">
+ <PatternLayout pattern="%d (%t) [%p - %l] %m%n" />
+ </Console>
+ </Appenders>
+ <Loggers>
+ <Logger name="org.apache.flume" level="DEBUG"/>
+ <Root level="DEBUG">
+ <AppenderRef ref="Console" />
+ </Root>
+ </Loggers>
+</Configuration>
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
new file mode 100644
index 0000000..553aaa6
--- /dev/null
+++ b/pom.xml
@@ -0,0 +1,277 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements. See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You 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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+
+ <parent>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-parent</artifactId>
+ <version>1.10.0-SNAPSHOT</version>
+ </parent>
+
+ <modelVersion>4.0.0</modelVersion>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-search</artifactId>
+ <version>1.0.0-SNAPSHOT</version>
+ <packaging>pom</packaging>
+
+ <name>Apache Flume Search Components</name>
+
+ <properties>
+ <!-- Set default encoding to UTF-8 to remove maven complaints -->
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+
+ <!-- Java compiler configuration -->
+ <sourceJavaVersion>1.8</sourceJavaVersion>
+ <targetJavaVersion>1.8</targetJavaVersion>
+
+ <elasticsearch.version>0.90.1</elasticsearch.version>
+ <flume.version>1.10.0-SNAPSHOT</flume.version>
+ </properties>
+
+ <modules>
+ <module>flume-elasticsearch-sink</module>
+ </modules>
+
+ <inceptionYear>2009</inceptionYear>
+
+ <issueManagement>
+ <system>JIRA</system>
+ <url>https://issues.apache.org/jira/browse/FLUME</url>
+ </issueManagement>
+
+ <licenses>
+ <license>
+ <name>The Apache Software License, Version 2.0</name>
+ <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+ </license>
+ </licenses>
+
+ <mailingLists>
+ <mailingList>
+ <archive>http://mail-archives.apache.org/mod_mbox/flume-user/</archive>
+ <name>Flume User List</name>
+ <post>user@flume.apache.org</post>
+ <subscribe>user-subscribe@flume.apache.org</subscribe>
+ <unsubscribe>user-unsubscribe@flume.apache.org</unsubscribe>
+ </mailingList>
+ <mailingList>
+ <archive>http://mail-archives.apache.org/mod_mbox/flume-dev/</archive>
+ <name>Flume Developer List</name>
+ <post>dev@flume.apache.org</post>
+ <subscribe>dev-subscribe@flume.apache.org</subscribe>
+ <unsubscribe>dev-unsubscribe@flume.apache.org</unsubscribe>
+ </mailingList>
+ <mailingList>
+ <archive>http://mail-archives.apache.org/mod_mbox/flume-commits/</archive>
+ <name>Flume Commits</name>
+ <post>commits@flume.apache.org</post>
+ <subscribe>commits-subscribe@flume.apache.org</subscribe>
+ <unsubscribe>commits-unsubscribe@flume.apache.org</unsubscribe>
+ </mailingList>
+ </mailingLists>
+
+ <scm>
+ <url>https://git-wip-us.apache.org/repos/asf?p=flume.git;a=tree;h=refs/heads/trunk;hb=trunk</url>
+ <developerConnection>scm:git:https://git-wip-us.apache.org/repos/asf/flume.git</developerConnection>
+ <connection>scm:git:http://git-wip-us.apache.org/repos/asf/flume.git</connection>
+ </scm>
+
+ <ciManagement>
+ <system>jenkins</system>
+ <url>https://builds.apache.org/job/flume-trunk</url>
+ </ciManagement>
+
+ <developers>
+ <developer>
+ <name>Ralph Goers</name>
+ <id>rgoers</id>
+ <email>rgoers@apache.org</email>
+ <organization>Intuit</organization>
+ </developer>
+ </developers>
+
+ <organization>
+ <name>Apache Software Foundation</name>
+ <url>http://www.apache.org</url>
+ </organization>
+
+ <distributionManagement>
+ <repository>
+ <id>apache.staging.https</id>
+ <name>Apache Staging Repository</name>
+ <url>https://repository.apache.org/service/local/staging/deploy/maven2/</url>
+ </repository>
+ <site>
+ <id>apache.website</id>
+ <url>${siteUrlDeployment}</url>
+ <name>Flume Site</name>
+ </site>
+ </distributionManagement>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-checkstyle-plugin</artifactId>
+ <version>${mvn-checkstyle-plugin.version}</version>
+ <dependencies>
+ <!-- The build-support module adds the checkstyle config files to the classpath. -->
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>build-support</artifactId>
+ <version>${flume.version}</version>
+ </dependency>
+ <!-- Newer versions of puppycrawl checkstyle have more features. Pull in a more recent
+ version than is specified in the maven-checkstyle-plugin pom file. -->
+ <dependency>
+ <groupId>com.puppycrawl.tools</groupId>
+ <artifactId>checkstyle</artifactId>
+ <version>${checkstyle.tool.version}</version>
+ </dependency>
+ </dependencies>
+ <executions>
+ <execution>
+ <id>verify.checkstyle</id>
+ <phase>verify</phase>
+ <goals>
+ <goal>check</goal>
+ </goals>
+ </execution>
+ </executions>
+ <configuration>
+ <configLocation>config/checkstyle/checkstyle.xml</configLocation>
+ <suppressionsLocation>config/checkstyle/checkstyle-suppressions.xml</suppressionsLocation>
+ <suppressionsFileExpression>checkstyle.suppressions.file</suppressionsFileExpression>
+ <encoding>UTF-8</encoding>
+ <includeTestSourceDirectory>true</includeTestSourceDirectory>
+ <linkXRef>false</linkXRef>
+ </configuration>
+ </plugin>
+ </plugins>
+ </build>
+
+ <dependencyManagement>
+ <dependencies>
+
+ <dependency>
+ <groupId>org.elasticsearch</groupId>
+ <artifactId>elasticsearch</artifactId>
+ <version>${elasticsearch.version}</version>
+ <optional>true</optional>
+ </dependency>
+
+ <dependency>
+ <groupId>org.elasticsearch</groupId>
+ <artifactId>elasticsearch</artifactId>
+ <version>${elasticsearch.version}</version>
+ <classifier>tests</classifier>
+ <scope>test</scope>
+ </dependency>
+
+ <!-- internal module dependencies -->
+
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-ng-configuration</artifactId>
+ <version>${flume.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-ng-config-filter-api</artifactId>
+ <version>${flume.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-ng-auth</artifactId>
+ <version>${flume.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-ng-core</artifactId>
+ <version>${flume.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>build-support</artifactId>
+ <version>${flume.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-tools</artifactId>
+ <version>${flume.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-ng-node</artifactId>
+ <version>${flume.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-ng-sdk</artifactId>
+ <version>${flume.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.flume</groupId>
+ <artifactId>flume-ng-sdk</artifactId>
+ <version>${flume.version}</version>
+ <classifier>tests</classifier>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+ </dependencyManagement>
+
+ <reporting>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-project-info-reports-plugin</artifactId>
+ <reportSets>
+ <reportSet>
+ <reports>
+ <report>ci-management</report>
+ <report>distribution-management</report>
+ <report>team</report>
+ <report>mailing-lists</report>
+ <report>issue-management</report>
+ <report>licenses</report>
+ <report>scm</report>
+ </reports>
+ </reportSet>
+ </reportSets>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-checkstyle-plugin</artifactId>
+ </plugin>
+ <plugin>
+ <groupId>com.github.spotbugs</groupId>
+ <artifactId>spotbugs-maven-plugin</artifactId>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-pmd-plugin</artifactId>
+ </plugin>
+ </plugins>
+ </reporting>
+
+</project>