This is an automated email from the ASF dual-hosted git repository.
rgoers pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/logging-flume.git
The following commit(s) were added to refs/heads/trunk by this push:
new 1278b6ddd Remove sinks (#430)
1278b6ddd is described below
commit 1278b6ddd6ed1b4e513c0ea810be53584eec9b44
Author: Piotr P. Karwasz <[email protected]>
AuthorDate: Wed May 27 02:21:01 2026 +0200
Remove sinks (#430)
Removes sinks that were moved to their own repositories:
- https://github.com/apache/logging-flume-http
- https://github.com/apache/logging-flume-irc
---
flume-ng-sinks/flume-http-sink/pom.xml | 134 -------
.../java/org/apache/flume/sink/http/HttpSink.java | 415 ---------------------
.../org/apache/flume/sink/http/package-info.java | 23 --
.../org/apache/flume/sink/http/TestHttpSink.java | 353 ------------------
.../org/apache/flume/sink/http/TestHttpSinkIT.java | 300 ---------------
.../flume-http-sink/src/test/resources/log4j2.xml | 30 --
flume-ng-sinks/flume-irc-sink/pom.xml | 86 -----
.../java/org/apache/flume/sink/irc/IRCSink.java | 266 -------------
.../org/apache/flume/sink/irc/TestIRCSink.java | 166 ---------
flume-ng-sinks/pom.xml | 43 ---
10 files changed, 1816 deletions(-)
diff --git a/flume-ng-sinks/flume-http-sink/pom.xml
b/flume-ng-sinks/flume-http-sink/pom.xml
deleted file mode 100644
index d4ac3f52a..000000000
--- a/flume-ng-sinks/flume-http-sink/pom.xml
+++ /dev/null
@@ -1,134 +0,0 @@
-<?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-ng-sinks</artifactId>
- <groupId>org.apache.flume</groupId>
- <version>2.0.0-SNAPSHOT</version>
- </parent>
-
- <groupId>org.apache.flume.flume-ng-sinks</groupId>
- <artifactId>flume-http-sink</artifactId>
- <name>Flume HTTP/S Sink</name>
-
- <properties>
- <!-- TODO fix spotbugs/pmd violations -->
- <spotbugs.maxAllowedViolations>1</spotbugs.maxAllowedViolations>
- <pmd.maxAllowedViolations>1</pmd.maxAllowedViolations>
- <module.name>org.apache.flume.sink.http</module.name>
- </properties>
-
- <dependencies>
-
- <dependency>
- <groupId>org.apache.flume</groupId>
- <artifactId>flume-ng-sdk</artifactId>
- <exclusions>
- <exclusion>
- <groupId>org.apache.thrift</groupId>
- <artifactId>libthrift</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flume</groupId>
- <artifactId>flume-ng-configuration</artifactId>
- <exclusions>
- <exclusion>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flume</groupId>
- <artifactId>flume-ng-core</artifactId>
- <exclusions>
- <exclusion>
- <groupId>org.apache.thrift</groupId>
- <artifactId>libthrift</artifactId>
- </exclusion>
- <exclusion>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
-
- <dependency>
- <groupId>org.slf4j</groupId>
- <artifactId>slf4j-api</artifactId>
- </dependency>
-
- <dependency>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- <version>${guava.version}</version>
- </dependency>
-
- <dependency>
- <groupId>commons-lang</groupId>
- <artifactId>commons-lang</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>org.mockito</groupId>
- <artifactId>mockito-all</artifactId>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>com.github.tomakehurst</groupId>
- <artifactId>wiremock</artifactId>
- <version>${wiremock.version}</version>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.apache.httpcomponents</groupId>
- <artifactId>httpclient</artifactId>
- <scope>test</scope>
- </dependency>
-
- </dependencies>
-
-</project>
diff --git
a/flume-ng-sinks/flume-http-sink/src/main/java/org/apache/flume/sink/http/HttpSink.java
b/flume-ng-sinks/flume-http-sink/src/main/java/org/apache/flume/sink/http/HttpSink.java
deleted file mode 100644
index d4d7ad9b3..000000000
---
a/flume-ng-sinks/flume-http-sink/src/main/java/org/apache/flume/sink/http/HttpSink.java
+++ /dev/null
@@ -1,415 +0,0 @@
-/*
- * 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.http;
-
-import org.apache.flume.Channel;
-import org.apache.flume.Context;
-import org.apache.flume.Event;
-import org.apache.flume.EventDeliveryException;
-import org.apache.flume.Transaction;
-import org.apache.flume.conf.Configurable;
-import org.apache.flume.instrumentation.SinkCounter;
-import org.apache.flume.sink.AbstractSink;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
-import java.io.OutputStream;
-import java.net.HttpURLConnection;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.util.HashMap;
-import java.util.Map;
-
-/**
- * Implementation of an HTTP sink. Events are POSTed to an HTTP / HTTPS
- * endpoint. The error handling behaviour is configurable, and can respond
- * differently depending on the response status returned by the endpoint.
- *
- * Rollback of the Flume transaction, and backoff can be specified globally,
- * then overridden for ranges (or individual) status codes.
- */
-public class HttpSink extends AbstractSink implements Configurable {
-
- /** Class logger. */
- private static final Logger LOG = LoggerFactory.getLogger(HttpSink.class);
-
- /** Lowest valid HTTP status code. */
- private static final int HTTP_STATUS_CONTINUE = 100;
-
- /** Default setting for the connection timeout when calling endpoint. */
- private static final int DEFAULT_CONNECT_TIMEOUT = 5000;
-
- /** Default setting for the request timeout when calling endpoint. */
- private static final int DEFAULT_REQUEST_TIMEOUT = 5000;
-
- /** Default setting for the HTTP content type header. */
- private static final String DEFAULT_CONTENT_TYPE = "text/plain";
-
- /** Default setting for the HTTP accept header. */
- private static final String DEFAULT_ACCEPT_HEADER = "text/plain";
-
- /** Endpoint URL to POST events to. */
- private URL endpointUrl;
-
- /** Counter used to monitor event throughput. */
- private SinkCounter sinkCounter;
-
- /** Actual connection timeout value in use. */
- private int connectTimeout = DEFAULT_CONNECT_TIMEOUT;
-
- /** Actual request timeout value in use. */
- private int requestTimeout = DEFAULT_REQUEST_TIMEOUT;
-
- /** Actual content type header value in use. */
- private String contentTypeHeader = DEFAULT_CONTENT_TYPE;
-
- /** Actual accept header value in use. */
- private String acceptHeader = DEFAULT_ACCEPT_HEADER;
-
- /** Backoff value to use if a specific override is not defined. */
- private boolean defaultBackoff;
-
- /** Rollback value to use if a specific override is not defined. */
- private boolean defaultRollback;
-
- /** Increment metrics value to use if a specific override is not defined. */
- private boolean defaultIncrementMetrics;
-
- /**
- * Holds all overrides for backoff. The key is a string of the format "500"
or
- * "5XX", and the value is the backoff value to use for the individual code,
- * or code range.
- */
- private HashMap<String, Boolean> backoffOverrides = new HashMap<>();
-
- /**
- * Holds all overrides for rollback. The key is a string of the format "500"
- * or "5XX", and the value is the rollback value to use for the individual
- * code, or code range.
- */
- private HashMap<String, Boolean> rollbackOverrides = new HashMap<>();
-
- /**
- * Holds all overrides for increment metrics. The key is a string of the
- * format "500" or "5XX", and the value is the increment metrics value to use
- * for the individual code, or code range.
- */
- private HashMap<String, Boolean> incrementMetricsOverrides = new HashMap<>();
-
- /** Used to create HTTP connections to the endpoint. */
- private ConnectionBuilder connectionBuilder;
-
- @Override
- public final void configure(final Context context) {
- String configuredEndpoint = context.getString("endpoint", "");
- LOG.info("Read endpoint URL from configuration : " + configuredEndpoint);
-
- try {
- endpointUrl = new URL(configuredEndpoint);
- } catch (MalformedURLException e) {
- throw new IllegalArgumentException("Endpoint URL invalid", e);
- }
-
- connectTimeout = context.getInteger("connectTimeout",
- DEFAULT_CONNECT_TIMEOUT);
-
- if (connectTimeout <= 0) {
- throw new IllegalArgumentException(
- "Connect timeout must be a non-zero and positive");
- }
- LOG.info("Using connect timeout : " + connectTimeout);
-
- requestTimeout = context.getInteger("requestTimeout",
- DEFAULT_REQUEST_TIMEOUT);
-
- if (requestTimeout <= 0) {
- throw new IllegalArgumentException(
- "Request timeout must be a non-zero and positive");
- }
- LOG.info("Using request timeout : " + requestTimeout);
-
- acceptHeader = context.getString("acceptHeader", DEFAULT_ACCEPT_HEADER);
- LOG.info("Using Accept header value : " + acceptHeader);
-
- contentTypeHeader = context.getString("contentTypeHeader",
- DEFAULT_CONTENT_TYPE);
- LOG.info("Using Content-Type header value : " + contentTypeHeader);
-
- defaultBackoff = context.getBoolean("defaultBackoff", true);
- LOG.info("Channel backoff by default is " + defaultBackoff);
-
- defaultRollback = context.getBoolean("defaultRollback", true);
- LOG.info("Transaction rollback by default is " + defaultRollback);
-
- defaultIncrementMetrics = context.getBoolean("defaultIncrementMetrics",
- false);
- LOG.info("Incrementing metrics by default is " + defaultIncrementMetrics);
-
- parseConfigOverrides("backoff", context, backoffOverrides);
- parseConfigOverrides("rollback", context, rollbackOverrides);
- parseConfigOverrides("incrementMetrics", context,
- incrementMetricsOverrides);
-
- if (this.sinkCounter == null) {
- this.sinkCounter = new SinkCounter(this.getName());
- }
-
- connectionBuilder = new ConnectionBuilder();
- }
-
- @Override
- public final void start() {
- LOG.info("Starting HttpSink");
- sinkCounter.start();
- }
-
- @Override
- public final void stop() {
- LOG.info("Stopping HttpSink");
- sinkCounter.stop();
- }
-
- @Override
- public final Status process() throws EventDeliveryException {
- Status status = null;
- OutputStream outputStream = null;
-
- Channel ch = getChannel();
- Transaction txn = ch.getTransaction();
- txn.begin();
-
- try {
- Event event = ch.take();
-
- byte[] eventBody = null;
- if (event != null) {
- eventBody = event.getBody();
- }
-
- if (eventBody != null && eventBody.length > 0) {
- sinkCounter.incrementEventDrainAttemptCount();
- LOG.debug("Sending request : " + new String(event.getBody()));
-
- try {
- HttpURLConnection connection = connectionBuilder.getConnection();
-
- outputStream = connection.getOutputStream();
- outputStream.write(eventBody);
- outputStream.flush();
- outputStream.close();
-
- int httpStatusCode = connection.getResponseCode();
- LOG.debug("Got status code : " + httpStatusCode);
-
- if (httpStatusCode < HttpURLConnection.HTTP_BAD_REQUEST) {
- connection.getInputStream().close();
- } else {
- LOG.debug("bad request");
- connection.getErrorStream().close();
- }
- LOG.debug("Response processed and closed");
-
- if (httpStatusCode >= HTTP_STATUS_CONTINUE) {
- String httpStatusString = String.valueOf(httpStatusCode);
-
- boolean shouldRollback = findOverrideValue(httpStatusString,
- rollbackOverrides, defaultRollback);
-
- if (shouldRollback) {
- txn.rollback();
- } else {
- txn.commit();
- }
-
- boolean shouldBackoff = findOverrideValue(httpStatusString,
- backoffOverrides, defaultBackoff);
-
- if (shouldBackoff) {
- status = Status.BACKOFF;
- } else {
- status = Status.READY;
- }
-
- boolean shouldIncrementMetrics =
findOverrideValue(httpStatusString,
- incrementMetricsOverrides, defaultIncrementMetrics);
-
- if (shouldIncrementMetrics) {
- sinkCounter.incrementEventDrainSuccessCount();
- }
-
- if (shouldRollback) {
- if (shouldBackoff) {
- LOG.info(String.format("Got status code %d from HTTP server."
- + " Rolled back event and backed off.", httpStatusCode));
- } else {
- LOG.info(String.format("Got status code %d from HTTP server."
- + " Rolled back event for retry.", httpStatusCode));
- }
- }
- } else {
- txn.rollback();
- status = Status.BACKOFF;
-
- LOG.warn("Malformed response returned from server, retrying");
- }
-
- } catch (IOException e) {
- txn.rollback();
- status = Status.BACKOFF;
-
- LOG.error("Error opening connection, or request timed out", e);
- sinkCounter.incrementEventWriteFail();
- }
-
- } else {
- txn.commit();
- status = Status.BACKOFF;
-
- LOG.warn("Processed empty event");
- }
-
- } catch (Throwable t) {
- txn.rollback();
- status = Status.BACKOFF;
-
- LOG.error("Error sending HTTP request, retrying", t);
- sinkCounter.incrementEventWriteOrChannelFail(t);
-
- // re-throw all Errors
- if (t instanceof Error) {
- throw (Error) t;
- }
-
- } finally {
- txn.close();
-
- if (outputStream != null) {
- try {
- outputStream.close();
- } catch (IOException e) {
- // ignore errors
- }
- }
- }
-
- return status;
- }
-
- /**
- * Reads a set of override values from the context configuration and stores
- * the results in the Map provided.
- *
- * @param propertyName the prefix of the config property names
- * @param context the context to use to read config properties
- * @param override the override Map to store results in
- */
- private void parseConfigOverrides(final String propertyName,
- final Context context,
- final Map<String, Boolean> override) {
-
- Map<String, String> config = context.getSubProperties(
- propertyName + ".");
-
- if (config != null) {
- for (Map.Entry<String, String> value : config.entrySet()) {
- LOG.info(String.format("Read %s value for status code %s as %s",
- propertyName, value.getKey(), value.getValue()));
-
- if (override.containsKey(value.getKey())) {
- LOG.warn(String.format("Ignoring duplicate config value for %s.%s",
- propertyName, value.getKey()));
- } else {
- override.put(value.getKey(), Boolean.valueOf(value.getValue()));
- }
- }
- }
- }
-
- /**
- * Queries the specified override map to find the most appropriate value. The
- * most specific match is found.
- *
- * @param statusCode the String representation of the HTTP status code
- * @param overrides the map of status code overrides
- * @param defaultValue the default value to use if no override is configured
- *
- * @return the value of the most specific match to the given status code
- */
- private boolean findOverrideValue(final String statusCode,
- final HashMap<String, Boolean> overrides,
- final boolean defaultValue) {
-
- Boolean overrideValue = overrides.get(statusCode);
- if (overrideValue == null) {
- overrideValue = overrides.get(statusCode.substring(0, 1) + "XX");
- if (overrideValue == null) {
- overrideValue = defaultValue;
- }
- }
- return overrideValue;
- }
-
- /**
- * Update the connection builder.
- *
- * @param builder the new value
- */
- final void setConnectionBuilder(final ConnectionBuilder builder) {
- this.connectionBuilder = builder;
- }
-
- /**
- * Update the sinkCounter.
- *
- * @param newSinkCounter the new value
- */
- final void setSinkCounter(final SinkCounter newSinkCounter) {
- this.sinkCounter = newSinkCounter;
- }
-
- /**
- * Class used to allow extending the connection building functionality.
- */
- class ConnectionBuilder {
-
- /**
- * Creates an HTTP connection to the configured endpoint address. This
- * connection is setup for a POST request, and uses the content type and
- * accept header values in the configuration.
- *
- * @return the connection object
- * @throws IOException on any connection error
- */
- public HttpURLConnection getConnection() throws IOException {
- HttpURLConnection connection = (HttpURLConnection)
- endpointUrl.openConnection();
-
- connection.setRequestMethod("POST");
- connection.setRequestProperty("Content-Type", contentTypeHeader);
- connection.setRequestProperty("Accept", acceptHeader);
- connection.setConnectTimeout(connectTimeout);
- connection.setReadTimeout(requestTimeout);
- connection.setDoOutput(true);
- connection.setDoInput(true);
- connection.connect();
- return connection;
- }
- }
-}
diff --git
a/flume-ng-sinks/flume-http-sink/src/main/java/org/apache/flume/sink/http/package-info.java
b/flume-ng-sinks/flume-http-sink/src/main/java/org/apache/flume/sink/http/package-info.java
deleted file mode 100644
index 5b88f01ce..000000000
---
a/flume-ng-sinks/flume-http-sink/src/main/java/org/apache/flume/sink/http/package-info.java
+++ /dev/null
@@ -1,23 +0,0 @@
-/*
- * 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.
- */
-/**
- * This package provides an HTTP sink for Flume so that events can be sent out
- * to a target HTTP endpoint.
- */
-package org.apache.flume.sink.http;
diff --git
a/flume-ng-sinks/flume-http-sink/src/test/java/org/apache/flume/sink/http/TestHttpSink.java
b/flume-ng-sinks/flume-http-sink/src/test/java/org/apache/flume/sink/http/TestHttpSink.java
deleted file mode 100644
index d8fcde2ca..000000000
---
a/flume-ng-sinks/flume-http-sink/src/test/java/org/apache/flume/sink/http/TestHttpSink.java
+++ /dev/null
@@ -1,353 +0,0 @@
-/*
- * 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.http;
-
-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.instrumentation.SinkCounter;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.mockito.Mock;
-import org.mockito.Mockito;
-import org.mockito.runners.MockitoJUnitRunner;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.OutputStream;
-import java.net.HttpURLConnection;
-
-import static org.mockito.Mockito.eq;
-import static org.mockito.Mockito.inOrder;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
-@RunWith(MockitoJUnitRunner.class)
-public class TestHttpSink {
-
- private static final Integer DEFAULT_REQUEST_TIMEOUT = 5000;
- private static final Integer DEFAULT_CONNECT_TIMEOUT = 5000;
- private static final String DEFAULT_ACCEPT_HEADER = "text/plain";
- private static final String DEFAULT_CONTENT_TYPE_HEADER = "text/plain";
-
- @Mock
- private SinkCounter sinkCounter;
-
- @Mock
- private Context configContext;
-
- @Mock
- private Channel channel;
-
- @Mock
- private Transaction transaction;
-
- @Mock
- private Event event;
-
- @Mock
- private HttpURLConnection httpURLConnection;
-
- @Mock
- private OutputStream outputStream;
-
- @Mock
- private InputStream inputStream;
-
- @Test
- public void ensureAllConfigurationOptionsRead() {
- whenDefaultStringConfig();
- whenDefaultBooleanConfig();
- when(configContext.getInteger(eq("connectTimeout"),
Mockito.anyInt())).thenReturn(1000);
- when(configContext.getInteger(eq("requestTimeout"),
Mockito.anyInt())).thenReturn(1000);
-
- new HttpSink().configure(configContext);
-
- verify(configContext).getString("endpoint", "");
- verify(configContext).getInteger(eq("connectTimeout"), Mockito.anyInt());
- verify(configContext).getInteger(eq("requestTimeout"), Mockito.anyInt());
- verify(configContext).getString(eq("acceptHeader"), Mockito.anyString());
- verify(configContext).getString(eq("contentTypeHeader"),
Mockito.anyString());
- verify(configContext).getBoolean("defaultBackoff", true);
- verify(configContext).getBoolean("defaultRollback", true);
- verify(configContext).getBoolean("defaultIncrementMetrics", false);
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void ensureExceptionIfEndpointUrlEmpty() {
- when(configContext.getString("endpoint", "")).thenReturn("");
- new HttpSink().configure(configContext);
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void ensureExceptionIfEndpointUrlInvalid() {
- when(configContext.getString("endpoint", "")).thenReturn("invalid url");
- new HttpSink().configure(configContext);
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void ensureExceptionIfConnectTimeoutNegative() {
- whenDefaultStringConfig();
- when(configContext.getInteger("connectTimeout", 1000)).thenReturn(-1000);
- when(configContext.getInteger(eq("requestTimeout"),
Mockito.anyInt())).thenReturn(1000);
- new HttpSink().configure(configContext);
- }
-
- @Test
- public void ensureDefaultConnectTimeoutCorrect() {
- whenDefaultStringConfig();
- when(configContext.getInteger("connectTimeout",
DEFAULT_CONNECT_TIMEOUT)).thenReturn(1000);
- when(configContext.getInteger(eq("requestTimeout"),
Mockito.anyInt())).thenReturn(1000);
- new HttpSink().configure(configContext);
- verify(configContext).getInteger("connectTimeout",
DEFAULT_CONNECT_TIMEOUT);
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void ensureExceptionIfRequestTimeoutNegative() {
- whenDefaultStringConfig();
- when(configContext.getInteger("requestTimeout", 1000)).thenReturn(-1000);
- when(configContext.getInteger(eq("connectTimeout"),
Mockito.anyInt())).thenReturn(1000);
- new HttpSink().configure(configContext);
- }
-
- @Test
- public void ensureDefaultRequestTimeoutCorrect() {
- whenDefaultStringConfig();
- when(configContext.getInteger("requestTimeout",
DEFAULT_REQUEST_TIMEOUT)).thenReturn(1000);
- when(configContext.getInteger(eq("connectTimeout"),
Mockito.anyInt())).thenReturn(1000);
- new HttpSink().configure(configContext);
- verify(configContext).getInteger("requestTimeout",
DEFAULT_REQUEST_TIMEOUT);
- }
-
- @Test
- public void ensureDefaultAcceptHeaderCorrect() {
- whenDefaultTimeouts();
- whenDefaultStringConfig();
- new HttpSink().configure(configContext);
- verify(configContext).getString("acceptHeader", DEFAULT_ACCEPT_HEADER);
- }
-
- @Test
- public void ensureDefaultContentTypeHeaderCorrect() {
- whenDefaultTimeouts();
- whenDefaultStringConfig();
- new HttpSink().configure(configContext);
- verify(configContext).getString("contentTypeHeader",
DEFAULT_CONTENT_TYPE_HEADER);
- }
-
- @Test
- public void ensureBackoffOnNullEvent() throws Exception {
- when(channel.take()).thenReturn(null);
- executeWithMocks(true);
- }
-
- @Test
- public void ensureBackoffOnNullEventBody() throws Exception {
- when(channel.take()).thenReturn(event);
- when(event.getBody()).thenReturn(null);
- executeWithMocks(true);
- }
-
- @Test
- public void ensureBackoffOnEmptyEvent() throws Exception {
- when(channel.take()).thenReturn(event);
- when(event.getBody()).thenReturn(new byte[]{});
- executeWithMocks(true);
- }
-
- @Test
- public void ensureRollbackBackoffAndIncrementMetricsIfConfigured() throws
Exception {
- when(channel.take()).thenReturn(event);
- when(event.getBody()).thenReturn("something".getBytes());
-
- Context context = new Context();
- context.put("defaultRollback", "true");
- context.put("defaultBackoff", "true");
- context.put("defaultIncrementMetrics", "true");
-
- executeWithMocks(false, Status.BACKOFF, true, true, context,
HttpURLConnection.HTTP_OK);
- }
-
- @Test
- public void ensureCommitReadyAndNoIncrementMetricsIfConfigured() throws
Exception {
- when(channel.take()).thenReturn(event);
- when(event.getBody()).thenReturn("something".getBytes());
-
- Context context = new Context();
- context.put("defaultRollback", "false");
- context.put("defaultBackoff", "false");
- context.put("defaultIncrementMetrics", "false");
-
- executeWithMocks(true, Status.READY, false, false, context,
HttpURLConnection.HTTP_OK);
- }
-
- @Test
- public void ensureSingleStatusConfigurationCorrectlyUsed() throws Exception {
- when(channel.take()).thenReturn(event);
- when(event.getBody()).thenReturn("something".getBytes());
-
- Context context = new Context();
- context.put("defaultRollback", "true");
- context.put("defaultBackoff", "true");
- context.put("defaultIncrementMetrics", "false");
- context.put("rollback.200", "false");
- context.put("backoff.200", "false");
- context.put("incrementMetrics.200", "true");
-
- executeWithMocks(true, Status.READY, true, true, context,
HttpURLConnection.HTTP_OK);
- }
-
- @Test
- public void testErrorCounter() throws Exception {
- RuntimeException exception = new RuntimeException("dummy");
- when(channel.take()).thenThrow(exception);
-
- Context context = new Context();
- context.put("defaultRollback", "false");
- context.put("defaultBackoff", "false");
- context.put("defaultIncrementMetrics", "false");
-
- executeWithMocks(false, Status.BACKOFF, false, false, context,
HttpURLConnection.HTTP_OK);
-
inOrder(sinkCounter).verify(sinkCounter).incrementEventWriteOrChannelFail(exception);
- }
-
- @Test
- public void ensureSingleErrorStatusConfigurationCorrectlyUsed() throws
Exception {
- when(channel.take()).thenReturn(event);
- when(event.getBody()).thenReturn("something".getBytes());
-
- Context context = new Context();
- context.put("defaultRollback", "true");
- context.put("defaultBackoff", "true");
- context.put("defaultIncrementMetrics", "false");
- context.put("rollback.401", "false");
- context.put("backoff.401", "false");
- context.put("incrementMetrics.401", "false");
-
- executeWithMocks(true, Status.READY, false, true, context,
HttpURLConnection.HTTP_UNAUTHORIZED);
- }
-
- @Test
- public void ensureGroupConfigurationCorrectlyUsed() throws Exception {
- when(channel.take()).thenReturn(event);
- when(event.getBody()).thenReturn("something".getBytes());
-
- Context context = new Context();
- context.put("defaultRollback", "true");
- context.put("defaultBackoff", "true");
- context.put("defaultIncrementMetrics", "false");
- context.put("rollback.2XX", "false");
- context.put("backoff.2XX", "false");
- context.put("incrementMetrics.2XX", "true");
-
- executeWithMocks(true, Status.READY, true, true, context,
HttpURLConnection.HTTP_OK);
- executeWithMocks(true, Status.READY, true, true, context,
HttpURLConnection.HTTP_NO_CONTENT);
- }
-
- @Test
- public void
ensureSingleStatusConfigurationOverridesGroupConfigurationCorrectly()
- throws Exception {
-
- when(channel.take()).thenReturn(event);
- when(event.getBody()).thenReturn("something".getBytes());
-
- Context context = new Context();
- context.put("rollback.2XX", "false");
- context.put("backoff.2XX", "false");
- context.put("incrementMetrics.2XX", "true");
- context.put("rollback.200", "true");
- context.put("backoff.200", "true");
- context.put("incrementMetrics.200", "false");
-
- executeWithMocks(true, Status.READY, true, true, context,
HttpURLConnection.HTTP_NO_CONTENT);
- executeWithMocks(false, Status.BACKOFF, false, true, context,
HttpURLConnection.HTTP_OK);
- }
-
- private void executeWithMocks(boolean commit) throws Exception {
- Context context = new Context();
- executeWithMocks(commit, Status.BACKOFF, false, false, context,
HttpURLConnection.HTTP_OK);
- }
-
- private void executeWithMocks(boolean expectedCommit, Status expectedStatus,
- boolean expectedIncrementSuccessMetrics,
- boolean expectedIncrementAttemptMetrics,
- Context context, int httpStatus)
- throws Exception {
-
- context.put("endpoint", "http://localhost:8080/endpoint";);
-
- HttpSink httpSink = new HttpSink();
- httpSink.configure(context);
- httpSink.setConnectionBuilder(httpSink.new ConnectionBuilder() {
- @Override
- public HttpURLConnection getConnection() throws IOException {
- return httpURLConnection;
- }
- });
- httpSink.setChannel(channel);
- httpSink.setSinkCounter(sinkCounter);
-
- when(channel.getTransaction()).thenReturn(transaction);
- when(httpURLConnection.getOutputStream()).thenReturn(outputStream);
- when(httpURLConnection.getInputStream()).thenReturn(inputStream);
- when(httpURLConnection.getErrorStream()).thenReturn(inputStream);
- when(httpURLConnection.getResponseCode()).thenReturn(httpStatus);
-
- Status actualStatus = httpSink.process();
-
- assert (actualStatus == expectedStatus);
-
- inOrder(transaction).verify(transaction).begin();
-
- if (expectedIncrementAttemptMetrics) {
-
inOrder(sinkCounter).verify(sinkCounter).incrementEventDrainAttemptCount();
- }
-
- if (expectedCommit) {
- inOrder(transaction).verify(transaction).commit();
- } else {
- inOrder(transaction).verify(transaction).rollback();
- }
-
- if (expectedIncrementSuccessMetrics) {
-
inOrder(sinkCounter).verify(sinkCounter).incrementEventDrainSuccessCount();
- }
-
- inOrder(transaction).verify(transaction).close();
- }
-
- private void whenDefaultStringConfig() {
- when(configContext.getString("endpoint",
"")).thenReturn("http://test.abc/";);
- when(configContext.getString("acceptHeader",
"")).thenReturn("test/accept");
- when(configContext.getString("contentTypeHeader",
"")).thenReturn("test/content");
- }
-
- private void whenDefaultBooleanConfig() {
- when(configContext.getBoolean("defaultBackoff", true)).thenReturn(true);
- when(configContext.getBoolean("defaultRollback", true)).thenReturn(true);
- when(configContext.getBoolean("defaultIncrementMetrics",
false)).thenReturn(true);
- }
-
- private void whenDefaultTimeouts() {
- when(configContext.getInteger(eq("requestTimeout"),
Mockito.anyInt())).thenReturn(1000);
- when(configContext.getInteger(eq("connectTimeout"),
Mockito.anyInt())).thenReturn(1000);
- }
-}
diff --git
a/flume-ng-sinks/flume-http-sink/src/test/java/org/apache/flume/sink/http/TestHttpSinkIT.java
b/flume-ng-sinks/flume-http-sink/src/test/java/org/apache/flume/sink/http/TestHttpSinkIT.java
deleted file mode 100644
index 9e2c356ec..000000000
---
a/flume-ng-sinks/flume-http-sink/src/test/java/org/apache/flume/sink/http/TestHttpSinkIT.java
+++ /dev/null
@@ -1,300 +0,0 @@
-/*
- * 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.http;
-
-import com.github.tomakehurst.wiremock.global.RequestDelaySpec;
-import com.github.tomakehurst.wiremock.http.Fault;
-import com.github.tomakehurst.wiremock.http.Request;
-import com.github.tomakehurst.wiremock.http.RequestListener;
-import com.github.tomakehurst.wiremock.http.Response;
-import com.github.tomakehurst.wiremock.junit.WireMockRule;
-import org.apache.flume.Context;
-import org.apache.flume.EventDeliveryException;
-import org.apache.flume.Sink;
-import org.apache.flume.Transaction;
-import org.apache.flume.channel.MemoryChannel;
-import org.apache.flume.event.SimpleEvent;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.mockito.runners.MockitoJUnitRunner;
-
-import java.io.IOException;
-import java.net.ServerSocket;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-
-import static com.github.tomakehurst.wiremock.client.WireMock.*;
-import static
com.github.tomakehurst.wiremock.core.WireMockConfiguration.wireMockConfig;
-import static com.github.tomakehurst.wiremock.stubbing.Scenario.STARTED;
-import static org.apache.flume.Sink.Status;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-
-/**
- * Runs a set of tests against a mocked HTTP endpoint.
- */
-@RunWith(MockitoJUnitRunner.class)
-public class TestHttpSinkIT {
-
- private static final int RESPONSE_TIMEOUT = 4000;
- private static final int CONNECT_TIMEOUT = 2500;
-
- private MemoryChannel channel;
-
- private HttpSink httpSink;
-
- private static int findFreePort() {
- try (ServerSocket socket = new ServerSocket(0)) {
- return socket.getLocalPort();
- } catch (IOException e) {
- throw new AssertionError("Can not find free port.", e);
- }
- }
-
- private final int port = findFreePort();
-
- @Before
- public void setupSink() {
- if (httpSink == null) {
- Context httpSinkContext = new Context();
- httpSinkContext.put("endpoint", "http://localhost:"; + port +
"/endpoint");
- httpSinkContext.put("requestTimeout", "2000");
- httpSinkContext.put("connectTimeout", "1500");
- httpSinkContext.put("acceptHeader", "application/json");
- httpSinkContext.put("contentTypeHeader", "application/json");
- httpSinkContext.put("backoff.200", "false");
- httpSinkContext.put("rollback.200", "false");
- httpSinkContext.put("backoff.401", "false");
- httpSinkContext.put("rollback.401", "false");
- httpSinkContext.put("incrementMetrics.200", "true");
-
- Context memoryChannelContext = new Context();
-
- channel = new MemoryChannel();
- channel.configure(memoryChannelContext);
- channel.start();
-
- httpSink = new HttpSink();
- httpSink.configure(httpSinkContext);
- httpSink.setChannel(channel);
- httpSink.start();
- }
- }
-
- @After
- public void waitForShutdown() throws InterruptedException {
- httpSink.stop();
- Thread.sleep(500);
- }
-
- @Rule
- public WireMockRule service = new WireMockRule(wireMockConfig().port(port));
-
- @Test
- public void ensureSuccessfulMessageDelivery() throws Exception {
- service.stubFor(post(urlEqualTo("/endpoint"))
- .withRequestBody(equalToJson(event("SUCCESS")))
- .willReturn(aResponse().withStatus(200)));
-
- addEventToChannel(event("SUCCESS"));
-
- service.verify(1, postRequestedFor(urlEqualTo("/endpoint"))
- .withRequestBody(equalToJson(event("SUCCESS"))));
- }
-
- @Test
- public void ensureEventsResentOn503Failure() throws Exception {
- String errorScenario = "Error Scenario";
-
- service.stubFor(post(urlEqualTo("/endpoint"))
- .inScenario(errorScenario)
- .whenScenarioStateIs(STARTED)
- .withRequestBody(equalToJson(event("TRANSIENT_ERROR")))
- .willReturn(aResponse().withStatus(503))
- .willSetStateTo("Error Sent"));
-
- service.stubFor(post(urlEqualTo("/endpoint"))
- .inScenario(errorScenario)
- .whenScenarioStateIs("Error Sent")
- .withRequestBody(equalToJson(event("TRANSIENT_ERROR")))
- .willReturn(aResponse().withStatus(200)));
-
- addEventToChannel(event("TRANSIENT_ERROR"), Status.BACKOFF);
- addEventToChannel(event("TRANSIENT_ERROR"), Status.READY);
-
- service.verify(2, postRequestedFor(urlEqualTo("/endpoint"))
- .withRequestBody(equalToJson(event("TRANSIENT_ERROR"))));
- }
-
- @Test
- public void ensureEventsNotResentOn401Failure() throws Exception {
- String errorScenario = "Error skip scenario";
-
- service.stubFor(post(urlEqualTo("/endpoint"))
- .inScenario(errorScenario)
- .whenScenarioStateIs(STARTED)
- .withRequestBody(equalToJson(event("UNAUTHORIZED REQUEST")))
- .willReturn(aResponse().withStatus(401)
- .withHeader("Content-Type", "text/plain")
- .withBody("Not allowed!"))
- .willSetStateTo("Error Sent"));
-
- service.stubFor(post(urlEqualTo("/endpoint"))
- .inScenario(errorScenario)
- .whenScenarioStateIs("Error Sent")
- .withRequestBody(equalToJson(event("NEXT EVENT")))
- .willReturn(aResponse().withStatus(200)));
-
- addEventToChannel(event("UNAUTHORIZED REQUEST"), Status.READY);
- addEventToChannel(event("NEXT EVENT"), Status.READY);
-
- service.verify(1, postRequestedFor(urlEqualTo("/endpoint"))
- .withRequestBody(equalToJson(event("UNAUTHORIZED REQUEST"))));
-
- service.verify(1, postRequestedFor(urlEqualTo("/endpoint"))
- .withRequestBody(equalToJson(event("NEXT EVENT"))));
-
- }
-
- @Test
- public void ensureEventsResentOnNetworkFailure() throws Exception {
- String errorScenario = "Error Scenario";
-
- service.stubFor(post(urlEqualTo("/endpoint"))
- .inScenario(errorScenario)
- .whenScenarioStateIs(STARTED)
- .withRequestBody(equalToJson(event("NETWORK_ERROR")))
- .willReturn(aResponse().withFault(Fault.RANDOM_DATA_THEN_CLOSE))
- .willSetStateTo("Error Sent"));
-
- service.stubFor(post(urlEqualTo("/endpoint"))
- .inScenario(errorScenario)
- .whenScenarioStateIs("Error Sent")
- .withRequestBody(equalToJson(event("NETWORK_ERROR")))
- .willReturn(aResponse().withStatus(200)));
-
- addEventToChannel(event("NETWORK_ERROR"), Status.BACKOFF);
- addEventToChannel(event("NETWORK_ERROR"), Status.READY);
-
- service.verify(2, postRequestedFor(urlEqualTo("/endpoint"))
- .withRequestBody(equalToJson(event("NETWORK_ERROR"))));
- }
-
- @Test
- public void ensureEventsResentOnConnectionTimeout() throws Exception {
- final CountDownLatch firstRequestReceived = new CountDownLatch(1);
-
- service.addSocketAcceptDelay(new RequestDelaySpec(CONNECT_TIMEOUT));
- service.addMockServiceRequestListener(new RequestListener() {
- @Override
- public void requestReceived(Request request, Response response) {
- service.addSocketAcceptDelay(new RequestDelaySpec(0));
- firstRequestReceived.countDown();
- }
- });
-
- service.stubFor(post(urlEqualTo("/endpoint"))
- .withRequestBody(equalToJson(event("SLOW_SOCKET")))
- .willReturn(aResponse().withStatus(200)));
-
- addEventToChannel(event("SLOW_SOCKET"), Status.BACKOFF);
-
- // wait until the socket is connected
- firstRequestReceived.await(2000, TimeUnit.MILLISECONDS);
-
- addEventToChannel(event("SLOW_SOCKET"), Status.READY);
-
- service.verify(2, postRequestedFor(urlEqualTo("/endpoint"))
- .withRequestBody(equalToJson(event("SLOW_SOCKET"))));
- }
-
- @Test
- public void ensureEventsResentOnRequestTimeout() throws Exception {
- String errorScenario = "Error Scenario";
-
- service.stubFor(post(urlEqualTo("/endpoint"))
- .inScenario(errorScenario)
- .whenScenarioStateIs(STARTED)
- .withRequestBody(equalToJson(event("SLOW_RESPONSE")))
-
.willReturn(aResponse().withFixedDelay(RESPONSE_TIMEOUT).withStatus(200))
- .willSetStateTo("Slow Response Sent"));
-
- service.stubFor(post(urlEqualTo("/endpoint"))
- .inScenario(errorScenario)
- .whenScenarioStateIs("Slow Response Sent")
- .withRequestBody(equalToJson(event("SLOW_RESPONSE")))
- .willReturn(aResponse().withStatus(200)));
-
- addEventToChannel(event("SLOW_RESPONSE"), Status.BACKOFF);
- addEventToChannel(event("SLOW_RESPONSE"), Status.READY);
-
- service.verify(2, postRequestedFor(urlEqualTo("/endpoint"))
- .withRequestBody(equalToJson(event("SLOW_RESPONSE"))));
- }
-
- @Test
- public void ensureHttpConnectionReusedForSuccessfulRequests() throws
Exception {
- // we should only get one delay when establishing a connection
- service.addSocketAcceptDelay(new RequestDelaySpec(1000));
-
- service.stubFor(post(urlEqualTo("/endpoint"))
- .withRequestBody(equalToJson(event("SUCCESS")))
- .willReturn(aResponse().withStatus(200)));
-
- long startTime = System.currentTimeMillis();
-
- addEventToChannel(event("SUCCESS"), Status.READY);
- addEventToChannel(event("SUCCESS"), Status.READY);
- addEventToChannel(event("SUCCESS"), Status.READY);
-
- long endTime = System.currentTimeMillis();
- assertTrue("Test should have completed faster", endTime - startTime <
2500);
-
- service.verify(3, postRequestedFor(urlEqualTo("/endpoint"))
- .withRequestBody(equalToJson(event("SUCCESS"))));
- }
-
- private void addEventToChannel(String line) throws EventDeliveryException {
- addEventToChannel(line, Status.READY);
- }
-
- private void addEventToChannel(String line, Status expectedStatus)
- throws EventDeliveryException {
-
- SimpleEvent event = new SimpleEvent();
- event.setBody(line.getBytes());
-
- Transaction channelTransaction = channel.getTransaction();
- channelTransaction.begin();
- channel.put(event);
- channelTransaction.commit();
- channelTransaction.close();
-
- Sink.Status status = httpSink.process();
-
- assertEquals(expectedStatus, status);
- }
-
- private String event(String id) {
- return "{'id':'" + id + "'}";
- }
-}
diff --git a/flume-ng-sinks/flume-http-sink/src/test/resources/log4j2.xml
b/flume-ng-sinks/flume-http-sink/src/test/resources/log4j2.xml
deleted file mode 100644
index 3b08de431..000000000
--- a/flume-ng-sinks/flume-http-sink/src/test/resources/log4j2.xml
+++ /dev/null
@@ -1,30 +0,0 @@
-<?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>
- <Root level="ALL">
- <AppenderRef ref="Console" />
- </Root>
- </Loggers>
-</Configuration>
\ No newline at end of file
diff --git a/flume-ng-sinks/flume-irc-sink/pom.xml
b/flume-ng-sinks/flume-irc-sink/pom.xml
deleted file mode 100644
index 53a5f4fe6..000000000
--- a/flume-ng-sinks/flume-irc-sink/pom.xml
+++ /dev/null
@@ -1,86 +0,0 @@
-<?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-ng-sinks</artifactId>
- <groupId>org.apache.flume</groupId>
- <version>2.0.0-SNAPSHOT</version>
- </parent>
-
- <groupId>org.apache.flume.flume-ng-sinks</groupId>
- <artifactId>flume-irc-sink</artifactId>
- <name>Flume NG IRC Sink</name>
-
- <properties>
- <!-- TODO fix spotbugs violations -->
- <spotbugs.maxAllowedViolations>1</spotbugs.maxAllowedViolations>
- <module.name>org.apache.flume.sink.irc</module.name>
- </properties>
-
- <dependencies>
-
- <dependency>
- <groupId>org.apache.flume</groupId>
- <artifactId>flume-ng-sdk</artifactId>
- </dependency>
-
- <dependency>
- <groupId>org.apache.flume</groupId>
- <artifactId>flume-ng-configuration</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.schwering</groupId>
- <artifactId>irclib</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>
-
- </dependencies>
-
-</project>
diff --git
a/flume-ng-sinks/flume-irc-sink/src/main/java/org/apache/flume/sink/irc/IRCSink.java
b/flume-ng-sinks/flume-irc-sink/src/main/java/org/apache/flume/sink/irc/IRCSink.java
deleted file mode 100644
index 5e7400c60..000000000
---
a/flume-ng-sinks/flume-irc-sink/src/main/java/org/apache/flume/sink/irc/IRCSink.java
+++ /dev/null
@@ -1,266 +0,0 @@
-/**
- * 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.irc;
-
-import java.io.IOException;
-
-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.Configurable;
-import org.apache.flume.exception.ChannelException;
-import org.apache.flume.sink.AbstractSink;
-import org.schwering.irc.lib.IRCConnection;
-import org.schwering.irc.lib.IRCEventListener;
-import org.schwering.irc.lib.IRCModeParser;
-import org.schwering.irc.lib.IRCUser;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.base.Preconditions;
-
-public class IRCSink extends AbstractSink implements Configurable {
-
- private static final Logger logger = LoggerFactory.getLogger(IRCSink.class);
-
- private static final int DEFAULT_PORT = 6667;
- private static final String DEFAULT_SPLIT_CHARS = "\n";
-
- private static final String IRC_CHANNEL_PREFIX = "#";
-
- private IRCConnection connection = null;
-
- private String hostname;
- private Integer port;
- private String nick;
- private String password;
- private String user;
- private String name;
- private String chan;
- private Boolean splitLines;
- private String splitChars;
-
- private CounterGroup counterGroup;
-
- public static class IRCConnectionListener implements IRCEventListener {
-
- public void onRegistered() {
- }
-
- public void onDisconnected() {
- logger.error("IRC sink disconnected");
- }
-
- public void onError(String msg) {
- logger.error("IRC sink error: {}", msg);
- }
-
- public void onError(int num, String msg) {
- logger.error("IRC sink error: {} - {}", num, msg);
- }
-
- public void onInvite(String chan, IRCUser u, String nickPass) {
- }
-
- public void onJoin(String chan, IRCUser u) {
- }
-
- public void onKick(String chan, IRCUser u, String nickPass, String msg) {
- }
-
- public void onMode(IRCUser u, String nickPass, String mode) {
- }
-
- public void onMode(String chan, IRCUser u, IRCModeParser mp) {
- }
-
- public void onNick(IRCUser u, String nickNew) {
- }
-
- public void onNotice(String target, IRCUser u, String msg) {
- }
-
- public void onPart(String chan, IRCUser u, String msg) {
- }
-
- public void onPrivmsg(String chan, IRCUser u, String msg) {
- }
-
- public void onQuit(IRCUser u, String msg) {
- }
-
- public void onReply(int num, String value, String msg) {
- }
-
- public void onTopic(String chan, IRCUser u, String topic) {
- }
-
- public void onPing(String p) {
- }
-
- public void unknown(String a, String b, String c, String d) {
- }
- }
-
- public IRCSink() {
- counterGroup = new CounterGroup();
- }
-
- public void configure(Context context) {
- hostname = context.getString("hostname");
- String portStr = context.getString("port");
- nick = context.getString("nick");
- password = context.getString("password");
- user = context.getString("user");
- name = context.getString("name");
- chan = context.getString("chan");
- splitLines = context.getBoolean("splitlines", false);
- splitChars = context.getString("splitchars");
-
- if (portStr != null) {
- port = Integer.parseInt(portStr);
- } else {
- port = DEFAULT_PORT;
- }
-
- if (splitChars == null) {
- splitChars = DEFAULT_SPLIT_CHARS;
- }
-
- Preconditions.checkState(hostname != null, "No hostname specified");
- Preconditions.checkState(nick != null, "No nick specified");
- Preconditions.checkState(chan != null, "No chan specified");
- }
-
- private void createConnection() throws IOException {
- if (connection == null) {
- logger.debug(
- "Creating new connection to hostname:{} port:{}",
- hostname, port);
- connection = new IRCConnection(hostname, new int[] { port },
- password, nick, user, name);
- connection.addIRCEventListener(new IRCConnectionListener());
- connection.setEncoding("UTF-8");
- connection.setPong(true);
- connection.setDaemon(false);
- connection.setColors(false);
- connection.connect();
- connection.send("join " + IRC_CHANNEL_PREFIX + chan);
- }
- }
-
- private void destroyConnection() {
- if (connection != null) {
- logger.debug("Destroying connection to: {}:{}", hostname, port);
- connection.close();
- }
-
- connection = null;
- }
-
- @Override
- public void start() {
- logger.info("IRC sink starting");
-
- try {
- createConnection();
- } catch (Exception e) {
- logger.error("Unable to create irc client using hostname:"
- + hostname + " port:" + port + ". Exception follows.", e);
-
- /* Try to prevent leaking resources. */
- destroyConnection();
-
- /* FIXME: Mark ourselves as failed. */
- return;
- }
-
- super.start();
-
- logger.debug("IRC sink {} started", this.getName());
- }
-
- @Override
- public void stop() {
- logger.info("IRC sink {} stopping", this.getName());
-
- destroyConnection();
-
- super.stop();
-
- logger.debug("IRC sink {} stopped. Metrics:{}", this.getName(),
counterGroup);
- }
-
- private void sendLine(Event event) {
- String body = new String(event.getBody());
-
- if (splitLines) {
- String[] lines = body.split(splitChars);
- for (String line: lines) {
- connection.doPrivmsg(IRC_CHANNEL_PREFIX + this.chan, line);
- }
- } else {
- connection.doPrivmsg(IRC_CHANNEL_PREFIX + this.chan, body);
- }
-
- }
-
- @Override
- public Status process() throws EventDeliveryException {
- Status status = Status.READY;
- Channel channel = getChannel();
- Transaction transaction = channel.getTransaction();
-
- try {
- transaction.begin();
- createConnection();
-
- Event event = channel.take();
-
- if (event == null) {
- counterGroup.incrementAndGet("event.empty");
- status = Status.BACKOFF;
- } else {
- sendLine(event);
- counterGroup.incrementAndGet("event.irc");
- }
-
- transaction.commit();
-
- } catch (ChannelException e) {
- transaction.rollback();
- logger.error(
- "Unable to get event from channel. Exception follows.", e);
- status = Status.BACKOFF;
- } catch (Exception e) {
- transaction.rollback();
- logger.error(
- "Unable to communicate with IRC server. Exception follows.",
- e);
- status = Status.BACKOFF;
- destroyConnection();
- } finally {
- transaction.close();
- }
-
- return status;
- }
-}
diff --git
a/flume-ng-sinks/flume-irc-sink/src/test/java/org/apache/flume/sink/irc/TestIRCSink.java
b/flume-ng-sinks/flume-irc-sink/src/test/java/org/apache/flume/sink/irc/TestIRCSink.java
deleted file mode 100644
index 32517d19f..000000000
---
a/flume-ng-sinks/flume-irc-sink/src/test/java/org/apache/flume/sink/irc/TestIRCSink.java
+++ /dev/null
@@ -1,166 +0,0 @@
-/**
- * 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.irc;
-
-import org.apache.commons.io.FileUtils;
-import org.apache.commons.io.IOUtils;
-import org.apache.flume.Channel;
-import org.apache.flume.Context;
-import org.apache.flume.Event;
-import org.apache.flume.EventDeliveryException;
-import org.apache.flume.Sink;
-import org.apache.flume.Transaction;
-import org.apache.flume.channel.MemoryChannel;
-import org.apache.flume.conf.Configurables;
-import org.apache.flume.event.EventBuilder;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Rule;
-import org.junit.Test;
-import org.junit.rules.TemporaryFolder;
-
-import java.io.File;
-import java.io.FileOutputStream;
-import java.io.IOException;
-import java.net.ServerSocket;
-import java.net.Socket;
-import java.util.List;
-import java.util.UUID;
-
-import static org.junit.Assert.fail;
-
-public class TestIRCSink {
-
- private File eventFile;
- int ircServerPort;
- DumbIRCServer dumbIRCServer;
- @Rule
- public TemporaryFolder folder = new TemporaryFolder();
-
- private static int findFreePort() throws IOException {
- ServerSocket socket = new ServerSocket(0);
- int port = socket.getLocalPort();
- socket.close();
- return port;
- }
-
- @Before
- public void setUp() throws IOException {
- ircServerPort = findFreePort();
- dumbIRCServer = new DumbIRCServer(ircServerPort);
- dumbIRCServer.start();
- eventFile = folder.newFile("eventFile.txt");
- }
-
- @After
- public void tearDown() throws Exception {
- dumbIRCServer.shutdownServer();
- }
-
- @Test
- public void testIRCSinkMissingSplitLineProperty() {
- Sink ircSink = new IRCSink();
- ircSink.setName("IRC Sink - " + UUID.randomUUID().toString());
- Context context = new Context();
- context.put("hostname", "localhost");
- context.put("port", String.valueOf(ircServerPort));
- context.put("nick", "flume");
- context.put("password", "flume");
- context.put("user", "flume");
- context.put("name", "flume-dev");
- context.put("chan", "flume");
- context.put("splitchars", "false");
- Configurables.configure(ircSink, context);
- Channel memoryChannel = new MemoryChannel();
- Configurables.configure(memoryChannel, context);
- ircSink.setChannel(memoryChannel);
- ircSink.start();
- Transaction txn = memoryChannel.getTransaction();
- txn.begin();
- Event event = EventBuilder.withBody("Dummy Event".getBytes());
- memoryChannel.put(event);
- txn.commit();
- txn.close();
- try {
- Sink.Status status = ircSink.process();
- if (status == Sink.Status.BACKOFF) {
- fail("Error occured");
- }
- } catch (EventDeliveryException eDelExcp) {
- // noop
- }
- }
-
- class DumbIRCServer extends Thread {
- int port;
- ServerSocket ss;
-
- public DumbIRCServer(int port) {
- this.port = port;
- }
-
- public void run() {
- try {
- ss = new ServerSocket(port);
- while (true) {
- try {
- Socket socket = ss.accept();
- process(socket);
- } catch (Exception ex) {
- /* noop */
- }
- }
- } catch (IOException e) {
- // noop
- }
- }
-
- public void shutdownServer() throws Exception {
- ss.close();
- }
-
- /**
- * Process the incoming request from IRC client
- *
- * @param socket IRC client connection socket
- * @throws IOException
- */
- private void process(Socket socket) throws IOException {
- FileOutputStream fileOutputStream =
FileUtils.openOutputStream(eventFile);
- List<String> input = IOUtils.readLines(socket.getInputStream());
- for (String next : input) {
- if (isPrivMessage(next)) {
- fileOutputStream.write(next.getBytes());
- fileOutputStream.write("\n".getBytes());
- }
- }
- fileOutputStream.close();
- socket.close();
- }
-
- /**
- * Checks if the message is Priv message
- *
- * @param input command received from IRC client
- * @return true, if command received is PrivMessage
- */
- private boolean isPrivMessage(String input) {
- return input.startsWith("PRIVMSG");
- }
- }
-}
\ No newline at end of file
diff --git a/flume-ng-sinks/pom.xml b/flume-ng-sinks/pom.xml
deleted file mode 100644
index 5f6240a9c..000000000
--- a/flume-ng-sinks/pom.xml
+++ /dev/null
@@ -1,43 +0,0 @@
-<?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-parent</artifactId>
- <groupId>org.apache.flume</groupId>
- <version>2.0.0-SNAPSHOT</version>
- <relativePath>../flume-parent/pom.xml</relativePath>
- </parent>
-
- <artifactId>flume-ng-sinks</artifactId>
- <name>Flume NG Sinks</name>
- <packaging>pom</packaging>
-
- <properties>
- <flume.sink.kudu.enabled>true</flume.sink.kudu.enabled>
- </properties>
-
- <modules>
- <module>flume-irc-sink</module>
- <module>flume-http-sink</module>
- </modules>
-
-</project>
