You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@logging.apache.org by rg...@apache.org on 2026/05/27 00:21:38 UTC
(logging-flume) branch trunk updated: Remove `jms` and `scribe` sources (#432)
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 eeb94a422 Remove `jms` and `scribe` sources (#432)
eeb94a422 is described below
commit eeb94a422bb7a45473090574153286ee3ad482dd
Author: Piotr P. Karwasz <pk...@apache.org>
AuthorDate: Wed May 27 02:21:33 2026 +0200
Remove `jms` and `scribe` sources (#432)
Removes sources that were moved to their own repositories:
- https://github.com/apache/logging-flume-jms
- https://github.com/apache/logging-flume-scribe
---
flume-ng-sources/flume-jms-source/pom.xml | 99 --
.../source/jms/DefaultJMSMessageConverter.java | 142 ---
.../flume/source/jms/InitialContextFactory.java | 30 -
.../flume/source/jms/JMSDestinationLocator.java | 23 -
.../flume/source/jms/JMSDestinationType.java | 23 -
.../flume/source/jms/JMSMessageConsumer.java | 213 ----
.../flume/source/jms/JMSMessageConverter.java | 51 -
.../org/apache/flume/source/jms/JMSSource.java | 402 --------
.../flume/source/jms/JMSSourceConfiguration.java | 69 --
.../source/jms/JMSMessageConsumerTestBase.java | 151 ---
.../source/jms/TestDefaultJMSMessageConverter.java | 189 ----
.../flume/source/jms/TestIntegrationActiveMQ.java | 334 -------
.../flume/source/jms/TestJMSMessageConsumer.java | 280 ------
.../org/apache/flume/source/jms/TestJMSSource.java | 384 -------
.../flume/source/jms/TestJMSSourceCreation.java | 49 -
flume-ng-sources/flume-scribe-source/pom.xml | 184 ----
.../org/apache/flume/source/scribe/LogEntry.java | 521 ----------
.../org/apache/flume/source/scribe/ResultCode.java | 63 --
.../org/apache/flume/source/scribe/Scribe.java | 1056 --------------------
.../apache/flume/source/scribe/ScribeSource.java | 185 ----
.../flume-scribe-source/src/main/thrift/aslv2 | 18 -
.../src/main/thrift/scribe-source.thrift | 34 -
.../flume/source/scribe/TestScribeSource.java | 165 ---
flume-ng-sources/pom.xml | 1 -
24 files changed, 4666 deletions(-)
diff --git a/flume-ng-sources/flume-jms-source/pom.xml b/flume-ng-sources/flume-jms-source/pom.xml
deleted file mode 100644
index f34b767b0..000000000
--- a/flume-ng-sources/flume-jms-source/pom.xml
+++ /dev/null
@@ -1,99 +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-sources</artifactId>
- <groupId>org.apache.flume</groupId>
- <version>2.0.0-SNAPSHOT</version>
- </parent>
-
- <groupId>org.apache.flume.flume-ng-sources</groupId>
- <artifactId>flume-jms-source</artifactId>
- <name>Flume JMS Source</name>
-
- <properties>
- <!-- TODO fix spotbugs/pmd violations -->
- <spotbugs.maxAllowedViolations>5</spotbugs.maxAllowedViolations>
- <pmd.maxAllowedViolations>4</pmd.maxAllowedViolations>
- <module.name>org.apache.flume.source.jms</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>junit</groupId>
- <artifactId>junit</artifactId>
- <scope>test</scope>
- </dependency>
-
- <dependency>
- <groupId>org.mockito</groupId>
- <artifactId>mockito-all</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.apache.geronimo.specs</groupId>
- <artifactId>geronimo-jms_1.1_spec</artifactId>
- </dependency>
-
- <dependency>
- <groupId>org.apache.activemq</groupId>
- <artifactId>activemq-core</artifactId>
- <scope>test</scope>
- </dependency>
-
- </dependencies>
-
-</project>
diff --git a/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/DefaultJMSMessageConverter.java b/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/DefaultJMSMessageConverter.java
deleted file mode 100644
index 003f591fd..000000000
--- a/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/DefaultJMSMessageConverter.java
+++ /dev/null
@@ -1,142 +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.source.jms;
-
-import java.io.ByteArrayOutputStream;
-import java.io.IOException;
-import java.io.ObjectOutput;
-import java.io.ObjectOutputStream;
-import java.nio.charset.Charset;
-import java.util.ArrayList;
-import java.util.Enumeration;
-import java.util.List;
-import java.util.Map;
-
-import javax.jms.BytesMessage;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.ObjectMessage;
-import javax.jms.TextMessage;
-
-import org.apache.flume.Context;
-import org.apache.flume.Event;
-import org.apache.flume.FlumeException;
-import org.apache.flume.event.SimpleEvent;
-
-/**
- * <p>Converts BytesMessage, TextMessage, and ObjectMessage
- * to a Flume Event. All Message Property names are added
- * as headers to the Event. The conversion of the body is
- * as follows:</p>
- *
- * <p><strong>BytesMessage:</strong> Body from message is
- * set as the body of the Event.</p>
- * <p><strong>TextMessage:</strong> String body converted to a byte
- * array byte getBytes(charset). Charset defaults to UTF-8 but can be
- * configured.</p>
- * <p><strong>ObjectMessage:</strong> Object is written to
- * an ByteArrayOutputStream wrapped by an ObjectOutputStream
- * and the resulting byte array is the body of the message.</p>
- */
-public class DefaultJMSMessageConverter implements JMSMessageConverter {
-
- private final Charset charset;
-
- private DefaultJMSMessageConverter(String charset) {
- this.charset = Charset.forName(charset);
- }
-
- public static class Builder implements JMSMessageConverter.Builder {
- @Override
- public JMSMessageConverter build(Context context) {
- return new DefaultJMSMessageConverter(context.getString(
- JMSSourceConfiguration.CONVERTER_CHARSET,
- JMSSourceConfiguration.CONVERTER_CHARSET_DEFAULT).trim());
- }
- }
-
- @Override
- public List<Event> convert(Message message) throws JMSException {
- Event event = new SimpleEvent();
- Map<String, String> headers = event.getHeaders();
- @SuppressWarnings("rawtypes")
- Enumeration propertyNames = message.getPropertyNames();
- while (propertyNames.hasMoreElements()) {
- String name = propertyNames.nextElement().toString();
- String value = message.getStringProperty(name);
- headers.put(name, value);
- }
- if (message instanceof BytesMessage) {
- BytesMessage bytesMessage = (BytesMessage)message;
- long length = bytesMessage.getBodyLength();
- if (length > 0L) {
- if (length > Integer.MAX_VALUE) {
- throw new JMSException("Unable to process message " + "of size "
- + length);
- }
- byte[] body = new byte[(int)length];
- int count = bytesMessage.readBytes(body);
- if (count != length) {
- throw new JMSException("Unable to read full message. " +
- "Read " + count + " of total " + length);
- }
- event.setBody(body);
- }
- } else if (message instanceof TextMessage) {
- TextMessage textMessage = (TextMessage)message;
- String text = textMessage.getText();
- if (text != null) {
- event.setBody(text.getBytes(charset));
- }
- } else if (message instanceof ObjectMessage) {
- ObjectMessage objectMessage = (ObjectMessage)message;
- Object object = objectMessage.getObject();
- if (object != null) {
- ByteArrayOutputStream bos = new ByteArrayOutputStream();
- ObjectOutput out = null;
- try {
- out = new ObjectOutputStream(bos);
- out.writeObject(object);
- event.setBody(bos.toByteArray());
- } catch (IOException e) {
- throw new FlumeException("Error serializing object", e);
- } finally {
- try {
- if (out != null) {
- out.close();
- }
- } catch (IOException e) {
- throw new FlumeException("Error closing ObjectOutputStream", e);
- }
- try {
- if (bos != null) {
- bos.close();
- }
- } catch (IOException e) {
- throw new FlumeException("Error closing ByteArrayOutputStream", e);
- }
- }
- }
-
- }
- List<Event> events = new ArrayList<Event>(1);
- events.add(event);
- return events;
- }
-}
diff --git a/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/InitialContextFactory.java b/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/InitialContextFactory.java
deleted file mode 100644
index 8874dd15a..000000000
--- a/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/InitialContextFactory.java
+++ /dev/null
@@ -1,30 +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.source.jms;
-
-import java.util.Properties;
-
-import javax.naming.InitialContext;
-import javax.naming.NamingException;
-
-public class InitialContextFactory {
-
- public InitialContext create(Properties properties) throws NamingException {
- return new InitialContext(properties);
- }
-}
diff --git a/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/JMSDestinationLocator.java b/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/JMSDestinationLocator.java
deleted file mode 100644
index c590c8e31..000000000
--- a/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/JMSDestinationLocator.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.
- */
-package org.apache.flume.source.jms;
-
-public enum JMSDestinationLocator {
- JNDI, CDI
-}
\ No newline at end of file
diff --git a/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/JMSDestinationType.java b/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/JMSDestinationType.java
deleted file mode 100644
index ac772168b..000000000
--- a/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/JMSDestinationType.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.
- */
-package org.apache.flume.source.jms;
-
-public enum JMSDestinationType {
- QUEUE(), TOPIC(),
-}
\ No newline at end of file
diff --git a/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/JMSMessageConsumer.java b/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/JMSMessageConsumer.java
deleted file mode 100644
index 3cbab0473..000000000
--- a/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/JMSMessageConsumer.java
+++ /dev/null
@@ -1,213 +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.source.jms;
-
-import com.google.common.base.Optional;
-import com.google.common.base.Preconditions;
-import org.apache.flume.Event;
-import org.apache.flume.FlumeException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import javax.jms.Connection;
-import javax.jms.ConnectionFactory;
-import javax.jms.Destination;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageConsumer;
-import javax.jms.Session;
-import javax.jms.Topic;
-import javax.naming.InitialContext;
-import javax.naming.NamingException;
-import java.util.ArrayList;
-import java.util.List;
-
-class JMSMessageConsumer {
- private static final Logger logger = LoggerFactory.getLogger(JMSMessageConsumer.class);
-
- private final int batchSize;
- private final long pollTimeout;
- private final JMSMessageConverter messageConverter;
-
- private final Connection connection;
- private final Session session;
- private final Destination destination;
- private final MessageConsumer messageConsumer;
-
- JMSMessageConsumer(InitialContext initialContext, ConnectionFactory connectionFactory,
- String destinationName, JMSDestinationLocator destinationLocator,
- JMSDestinationType destinationType, String messageSelector, int batchSize,
- long pollTimeout, JMSMessageConverter messageConverter,
- Optional<String> userName, Optional<String> password,
- Optional<String> clientId, boolean createDurableSubscription,
- String durableSubscriptionName) {
- this.batchSize = batchSize;
- this.pollTimeout = pollTimeout;
- this.messageConverter = messageConverter;
- Preconditions.checkArgument(batchSize > 0, "Batch size must be greater "
- + "than zero");
- Preconditions.checkArgument(pollTimeout >= 0, "Poll timeout cannot be " +
- "negative");
-
- try {
- try {
- if (userName.isPresent() && password.isPresent()) {
- connection = connectionFactory.createConnection(userName.get(), password.get());
- } else {
- connection = connectionFactory.createConnection();
- }
- if (clientId.isPresent()) {
- connection.setClientID(clientId.get());
- }
- connection.start();
- } catch (JMSException e) {
- throw new FlumeException("Could not create connection to broker", e);
- }
-
- try {
- session = connection.createSession(true, Session.SESSION_TRANSACTED);
- } catch (JMSException e) {
- throw new FlumeException("Could not create session", e);
- }
-
- try {
- if (destinationLocator.equals(JMSDestinationLocator.CDI)) {
- switch (destinationType) {
- case QUEUE:
- destination = session.createQueue(destinationName);
- break;
- case TOPIC:
- destination = session.createTopic(destinationName);
- break;
- default:
- throw new IllegalStateException(String.valueOf(destinationType));
- }
- } else {
- JMSSource.verifyContext(destinationName);
- destination = (Destination) initialContext.lookup(destinationName);
- }
- } catch (JMSException e) {
- throw new FlumeException("Could not create destination " + destinationName, e);
- } catch (NamingException e) {
- throw new FlumeException("Could not find destination " + destinationName, e);
- }
-
- try {
- if (createDurableSubscription) {
- messageConsumer = session.createDurableSubscriber(
- (Topic) destination, durableSubscriptionName,
- messageSelector.isEmpty() ? null : messageSelector, true);
- } else {
- messageConsumer = session.createConsumer(destination,
- messageSelector.isEmpty() ? null : messageSelector);
- }
- } catch (JMSException e) {
- throw new FlumeException("Could not create consumer", e);
- }
- String startupMsg = String.format("Connected to '%s' of type '%s' with " +
- "user '%s', batch size '%d', selector '%s' ", destinationName,
- destinationType, userName.isPresent() ? userName.get() : "null",
- batchSize, messageSelector.isEmpty() ? null : messageSelector);
- logger.info(startupMsg);
- } catch (Exception e) {
- close();
- throw e;
- }
- }
-
- List<Event> take() throws JMSException {
- List<Event> result = new ArrayList<Event>(batchSize);
- Message message;
- message = receive();
- if (message != null) {
- result.addAll(messageConverter.convert(message));
- int max = batchSize - 1;
- for (int i = 0; i < max; i++) {
- message = receiveNoWait();
- if (message == null) {
- break;
- }
- result.addAll(messageConverter.convert(message));
- }
- }
- if (logger.isDebugEnabled()) {
- logger.debug(String.format("Took batch of %s from %s", result.size(), destination));
- }
- return result;
- }
-
- private Message receive() throws JMSException {
- try {
- return messageConsumer.receive(pollTimeout);
- } catch (RuntimeException runtimeException) {
- JMSException jmsException = new JMSException("JMS provider has thrown runtime exception: "
- + runtimeException.getMessage());
- jmsException.setLinkedException(runtimeException);
- throw jmsException;
- }
- }
-
- private Message receiveNoWait() throws JMSException {
- try {
- return messageConsumer.receiveNoWait();
- } catch (RuntimeException runtimeException) {
- JMSException jmsException = new JMSException("JMS provider has thrown runtime exception: "
- + runtimeException.getMessage());
- jmsException.setLinkedException(runtimeException);
- throw jmsException;
- }
- }
-
- void commit() {
- try {
- session.commit();
- } catch (JMSException jmsException) {
- logger.warn("JMS Exception processing commit", jmsException);
- } catch (RuntimeException runtimeException) {
- logger.warn("Runtime Exception processing commit", runtimeException);
- }
- }
-
- void rollback() {
- try {
- session.rollback();
- } catch (JMSException jmsException) {
- logger.warn("JMS Exception processing rollback", jmsException);
- } catch (RuntimeException runtimeException) {
- logger.warn("Runtime Exception processing rollback", runtimeException);
- }
- }
-
- void close() {
- try {
- if (session != null) {
- session.close();
- }
- } catch (JMSException e) {
- logger.error("Could not destroy session", e);
- }
- try {
- if (connection != null) {
- connection.close();
- }
- } catch (JMSException e) {
- logger.error("Could not destroy connection", e);
- }
- }
-}
diff --git a/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/JMSMessageConverter.java b/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/JMSMessageConverter.java
deleted file mode 100644
index 31a15b874..000000000
--- a/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/JMSMessageConverter.java
+++ /dev/null
@@ -1,51 +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.source.jms;
-
-import java.util.List;
-
-import javax.jms.JMSException;
-import javax.jms.Message;
-
-import org.apache.flume.Context;
-import org.apache.flume.Event;
-import org.apache.flume.annotations.InterfaceAudience;
-import org.apache.flume.annotations.InterfaceStability;
-
-/**
- * Converts a JMS Message to an Event. It's recommended
- * that sub-classes define a static sub-class of the
- * inner Builder class to handle configuration. Alternatively,
- * the sub-class can the Configurable interface.
- *
- */
-@InterfaceAudience.Public
-@InterfaceStability.Stable
-public interface JMSMessageConverter {
-
- List<Event> convert(Message message) throws JMSException;
-
- /**
- * Implementors of JMSMessageConverter must either provide
- * a suitable builder or implement the Configurable interface.
- */
- interface Builder {
- JMSMessageConverter build(Context context);
- }
-}
diff --git a/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/JMSSource.java b/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/JMSSource.java
deleted file mode 100644
index e6f8a11d1..000000000
--- a/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/JMSSource.java
+++ /dev/null
@@ -1,402 +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.source.jms;
-
-import java.io.File;
-import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.Locale;
-import java.util.Properties;
-
-import javax.jms.ConnectionFactory;
-import javax.jms.JMSException;
-import javax.naming.InitialContext;
-import javax.naming.NamingException;
-
-import org.apache.commons.lang.StringUtils;
-import org.apache.flume.exception.ChannelException;
-import org.apache.flume.Context;
-import org.apache.flume.Event;
-import org.apache.flume.EventDeliveryException;
-import org.apache.flume.FlumeException;
-import org.apache.flume.annotations.InterfaceAudience;
-import org.apache.flume.annotations.InterfaceStability;
-import org.apache.flume.conf.BatchSizeSupported;
-import org.apache.flume.conf.Configurables;
-import org.apache.flume.instrumentation.SourceCounter;
-import org.apache.flume.source.AbstractPollableSource;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Charsets;
-import com.google.common.base.Optional;
-import com.google.common.base.Preconditions;
-import com.google.common.io.Files;
-
-@InterfaceAudience.Private
-@InterfaceStability.Unstable
-public class JMSSource extends AbstractPollableSource implements BatchSizeSupported {
- private static final Logger logger = LoggerFactory.getLogger(JMSSource.class);
- private static final String JAVA_SCHEME = "java";
- public static final String JNDI_ALLOWED_PROTOCOLS = "JndiAllowedProtocols";
-
- // setup by constructor
- private final InitialContextFactory initialContextFactory;
-
- // setup by configuration
- private ConnectionFactory connectionFactory;
- private int batchSize;
- private JMSMessageConverter converter;
- private JMSMessageConsumer consumer;
- private String initialContextFactoryName;
- private String providerUrl;
- private String destinationName;
- private JMSDestinationType destinationType;
- private JMSDestinationLocator destinationLocator;
- private String messageSelector;
- private Optional<String> userName;
- private Optional<String> password;
- private SourceCounter sourceCounter;
- private int errorThreshold;
- private long pollTimeout;
- private Optional<String> clientId;
- private boolean createDurableSubscription;
- private String durableSubscriptionName;
-
- private int jmsExceptionCounter;
- private InitialContext initialContext;
- private static List<String> allowedSchemes = getAllowedProtocols();
-
- public JMSSource() {
- this(new InitialContextFactory());
- }
-
- public JMSSource(InitialContextFactory initialContextFactory) {
- super();
- this.initialContextFactory = initialContextFactory;
- }
-
- private static List<String> getAllowedProtocols() {
- String allowed = System.getProperty(JNDI_ALLOWED_PROTOCOLS, null);
- if (allowed == null) {
- return Collections.singletonList(JAVA_SCHEME);
- } else {
- String[] items = allowed.split(",");
- List<String> schemes = new ArrayList<>();
- schemes.add(JAVA_SCHEME);
- for (String item : items) {
- if (!item.equals(JAVA_SCHEME)) {
- schemes.add(item.trim());
- }
- }
- return schemes;
- }
- }
-
- public static void verifyContext(String location) {
- try {
- String scheme = new URI(location).getScheme();
- if (scheme != null && !allowedSchemes.contains(scheme)) {
- throw new IllegalArgumentException("Invalid JNDI URI: " + location);
- }
- } catch (URISyntaxException ex) {
- logger.trace("{}} is not a valid URI", location);
- }
- }
-
- @Override
- protected void doConfigure(Context context) throws FlumeException {
- sourceCounter = new SourceCounter(getName());
-
- initialContextFactoryName = context.getString(
- JMSSourceConfiguration.INITIAL_CONTEXT_FACTORY, "").trim();
-
- providerUrl = context.getString(JMSSourceConfiguration.PROVIDER_URL, "").trim();
- verifyContext(providerUrl);
-
- destinationName = context.getString(JMSSourceConfiguration.DESTINATION_NAME, "").trim();
-
- String destinationTypeName = context.getString(
- JMSSourceConfiguration.DESTINATION_TYPE, "").trim().toUpperCase(Locale.ENGLISH);
-
- String destinationLocatorName = context.getString(
- JMSSourceConfiguration.DESTINATION_LOCATOR,
- JMSSourceConfiguration.DESTINATION_LOCATOR_DEFAULT)
- .trim().toUpperCase(Locale.ENGLISH);
-
- messageSelector = context.getString(
- JMSSourceConfiguration.MESSAGE_SELECTOR, "").trim();
-
- batchSize = context.getInteger(JMSSourceConfiguration.BATCH_SIZE,
- JMSSourceConfiguration.BATCH_SIZE_DEFAULT);
-
- errorThreshold = context.getInteger(JMSSourceConfiguration.ERROR_THRESHOLD,
- JMSSourceConfiguration.ERROR_THRESHOLD_DEFAULT);
-
- userName = Optional.fromNullable(context.getString(JMSSourceConfiguration.USERNAME));
-
- pollTimeout = context.getLong(JMSSourceConfiguration.POLL_TIMEOUT,
- JMSSourceConfiguration.POLL_TIMEOUT_DEFAULT);
-
- clientId = Optional.fromNullable(context.getString(JMSSourceConfiguration.CLIENT_ID));
-
- createDurableSubscription = context.getBoolean(
- JMSSourceConfiguration.CREATE_DURABLE_SUBSCRIPTION,
- JMSSourceConfiguration.DEFAULT_CREATE_DURABLE_SUBSCRIPTION);
- durableSubscriptionName = context.getString(
- JMSSourceConfiguration.DURABLE_SUBSCRIPTION_NAME,
- JMSSourceConfiguration.DEFAULT_DURABLE_SUBSCRIPTION_NAME);
-
- String passwordFile = context.getString(JMSSourceConfiguration.PASSWORD_FILE, "").trim();
-
- if (passwordFile.isEmpty()) {
- password = Optional.absent();
- } else {
- try {
- password = Optional.of(Files.toString(new File(passwordFile),
- Charsets.UTF_8).trim());
- } catch (IOException e) {
- throw new FlumeException(String.format(
- "Could not read password file %s", passwordFile), e);
- }
- }
-
- String converterClassName = context.getString(
- JMSSourceConfiguration.CONVERTER_TYPE,
- JMSSourceConfiguration.CONVERTER_TYPE_DEFAULT).trim();
- if (JMSSourceConfiguration.CONVERTER_TYPE_DEFAULT.equalsIgnoreCase(converterClassName)) {
- converterClassName = DefaultJMSMessageConverter.Builder.class.getName();
- }
- Context converterContext = new Context(context.getSubProperties(
- JMSSourceConfiguration.CONVERTER + "."));
- try {
- @SuppressWarnings("rawtypes")
- Class clazz = Class.forName(converterClassName);
- boolean isBuilder = JMSMessageConverter.Builder.class
- .isAssignableFrom(clazz);
- if (isBuilder) {
- JMSMessageConverter.Builder builder = (JMSMessageConverter.Builder)clazz.newInstance();
- converter = builder.build(converterContext);
- } else {
- Preconditions.checkState(JMSMessageConverter.class.isAssignableFrom(clazz),
- String.format("Class %s is not a subclass of JMSMessageConverter", clazz.getName()));
- converter = (JMSMessageConverter)clazz.newInstance();
- boolean configured = Configurables.configure(converter, converterContext);
- if (logger.isDebugEnabled()) {
- logger.debug(String.format("Attempted configuration of %s, result = %s",
- converterClassName, String.valueOf(configured)));
- }
- }
- } catch (Exception e) {
- throw new FlumeException(String.format(
- "Unable to create instance of converter %s", converterClassName), e);
- }
-
- String connectionFactoryName = context.getString(
- JMSSourceConfiguration.CONNECTION_FACTORY,
- JMSSourceConfiguration.CONNECTION_FACTORY_DEFAULT).trim();
- verifyContext(connectionFactoryName);
-
- assertNotEmpty(initialContextFactoryName, String.format(
- "Initial Context Factory is empty. This is specified by %s",
- JMSSourceConfiguration.INITIAL_CONTEXT_FACTORY));
-
- assertNotEmpty(providerUrl, String.format(
- "Provider URL is empty. This is specified by %s",
- JMSSourceConfiguration.PROVIDER_URL));
-
- assertNotEmpty(destinationName, String.format(
- "Destination Name is empty. This is specified by %s",
- JMSSourceConfiguration.DESTINATION_NAME));
-
- assertNotEmpty(destinationTypeName, String.format(
- "Destination Type is empty. This is specified by %s",
- JMSSourceConfiguration.DESTINATION_TYPE));
-
- try {
- destinationType = JMSDestinationType.valueOf(destinationTypeName);
- } catch (IllegalArgumentException e) {
- throw new FlumeException(String.format("Destination type '%s' is " +
- "invalid.", destinationTypeName), e);
- }
-
- if (createDurableSubscription) {
- if (JMSDestinationType.TOPIC != destinationType) {
- throw new FlumeException(String.format(
- "Only Destination type '%s' supports durable subscriptions.",
- JMSDestinationType.TOPIC.toString()));
- }
- if (!clientId.isPresent()) {
- throw new FlumeException(String.format(
- "You have to specify '%s' when using durable subscriptions.",
- JMSSourceConfiguration.CLIENT_ID));
- }
- if (StringUtils.isEmpty(durableSubscriptionName)) {
- throw new FlumeException(String.format("If '%s' is set to true, '%s' has to be specified.",
- JMSSourceConfiguration.CREATE_DURABLE_SUBSCRIPTION,
- JMSSourceConfiguration.DURABLE_SUBSCRIPTION_NAME));
- }
- } else if (!StringUtils.isEmpty(durableSubscriptionName)) {
- logger.warn(String.format("'%s' is set, but '%s' is false."
- + "If you want to create a durable subscription, set %s to true.",
- JMSSourceConfiguration.DURABLE_SUBSCRIPTION_NAME,
- JMSSourceConfiguration.CREATE_DURABLE_SUBSCRIPTION,
- JMSSourceConfiguration.CREATE_DURABLE_SUBSCRIPTION));
- }
-
- try {
- destinationLocator = JMSDestinationLocator.valueOf(destinationLocatorName);
- } catch (IllegalArgumentException e) {
- throw new FlumeException(String.format("Destination locator '%s' is " +
- "invalid.", destinationLocatorName), e);
- }
-
- Preconditions.checkArgument(batchSize > 0, "Batch size must be greater than 0");
-
- try {
- Properties contextProperties = new Properties();
- contextProperties.setProperty(
- javax.naming.Context.INITIAL_CONTEXT_FACTORY,
- initialContextFactoryName);
- contextProperties.setProperty(
- javax.naming.Context.PROVIDER_URL, providerUrl);
-
- // Provide properties for connecting via JNDI
- if (this.userName.isPresent()) {
- contextProperties.setProperty(javax.naming.Context.SECURITY_PRINCIPAL,
- this.userName.get());
- }
- if (this.password.isPresent()) {
- contextProperties.setProperty(javax.naming.Context.SECURITY_CREDENTIALS,
- this.password.get());
- }
-
- initialContext = initialContextFactory.create(contextProperties);
- } catch (NamingException e) {
- throw new FlumeException(String.format(
- "Could not create initial context %s provider %s",
- initialContextFactoryName, providerUrl), e);
- }
-
- try {
- connectionFactory = (ConnectionFactory) initialContext.lookup(connectionFactoryName);
- } catch (NamingException e) {
- throw new FlumeException("Could not lookup ConnectionFactory", e);
- }
- }
-
- private void assertNotEmpty(String arg, String msg) {
- Preconditions.checkArgument(!arg.isEmpty(), msg);
- }
-
- @Override
- protected synchronized Status doProcess() throws EventDeliveryException {
- boolean error = true;
- try {
- if (consumer == null) {
- consumer = createConsumer();
- }
- List<Event> events = consumer.take();
- int size = events.size();
- if (size == 0) {
- error = false;
- return Status.BACKOFF;
- }
- sourceCounter.incrementAppendBatchReceivedCount();
- sourceCounter.addToEventReceivedCount(size);
- getChannelProcessor().processEventBatch(events);
- error = false;
- sourceCounter.addToEventAcceptedCount(size);
- sourceCounter.incrementAppendBatchAcceptedCount();
- return Status.READY;
- } catch (ChannelException channelException) {
- logger.warn("Error appending event to channel. "
- + "Channel might be full. Consider increasing the channel "
- + "capacity or make sure the sinks perform faster.", channelException);
- sourceCounter.incrementChannelWriteFail();
- } catch (JMSException jmsException) {
- logger.warn("JMSException consuming events", jmsException);
- if (++jmsExceptionCounter > errorThreshold && consumer != null) {
- logger.warn("Exceeded JMSException threshold, closing consumer");
- sourceCounter.incrementEventReadFail();
- consumer.rollback();
- consumer.close();
- consumer = null;
- }
- } catch (Throwable throwable) {
- logger.error("Unexpected error processing events", throwable);
- sourceCounter.incrementEventReadFail();
- if (throwable instanceof Error) {
- throw (Error) throwable;
- }
- } finally {
- if (error) {
- if (consumer != null) {
- consumer.rollback();
- }
- } else {
- if (consumer != null) {
- consumer.commit();
- jmsExceptionCounter = 0;
- }
- }
- }
- return Status.BACKOFF;
- }
-
- @Override
- protected synchronized void doStart() {
- try {
- consumer = createConsumer();
- jmsExceptionCounter = 0;
- sourceCounter.start();
- } catch (JMSException e) {
- throw new FlumeException("Unable to create consumer", e);
- }
- }
-
- @Override
- protected synchronized void doStop() {
- if (consumer != null) {
- consumer.close();
- consumer = null;
- }
- sourceCounter.stop();
- }
-
- @VisibleForTesting
- JMSMessageConsumer createConsumer() throws JMSException {
- logger.info("Creating new consumer for " + destinationName);
- JMSMessageConsumer consumer = new JMSMessageConsumer(initialContext,
- connectionFactory, destinationName, destinationLocator, destinationType,
- messageSelector, batchSize, pollTimeout, converter, userName, password, clientId,
- createDurableSubscription, durableSubscriptionName);
- jmsExceptionCounter = 0;
- return consumer;
- }
-
- @Override
- public long getBatchSize() {
- return batchSize;
- }
-}
\ No newline at end of file
diff --git a/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/JMSSourceConfiguration.java b/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/JMSSourceConfiguration.java
deleted file mode 100644
index adb167bbd..000000000
--- a/flume-ng-sources/flume-jms-source/src/main/java/org/apache/flume/source/jms/JMSSourceConfiguration.java
+++ /dev/null
@@ -1,69 +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.source.jms;
-
-public class JMSSourceConfiguration {
-
- public static final String INITIAL_CONTEXT_FACTORY = "initialContextFactory";
-
- public static final String CONNECTION_FACTORY = "connectionFactory";
- public static final String CONNECTION_FACTORY_DEFAULT = "ConnectionFactory";
-
- public static final String PROVIDER_URL = "providerURL";
-
- public static final String DESTINATION_NAME = "destinationName";
-
- public static final String DESTINATION_TYPE = "destinationType";
- public static final String DESTINATION_LOCATOR = "destinationLocator";
- public static final String DESTINATION_LOCATOR_DEFAULT = "CDI";
-
- public static final String DESTINATION_TYPE_QUEUE = "queue";
- public static final String DESTINATION_TYPE_TOPIC = "topic";
-
- public static final String MESSAGE_SELECTOR = "messageSelector";
-
- public static final String CLIENT_ID = "clientId";
-
- public static final String USERNAME = "userName";
-
- public static final String PASSWORD_FILE = "passwordFile";
-
- public static final String BATCH_SIZE = "batchSize";
- public static final int BATCH_SIZE_DEFAULT = 100;
-
- public static final String ERROR_THRESHOLD = "errorThreshold";
- public static final int ERROR_THRESHOLD_DEFAULT = 10;
-
- public static final String POLL_TIMEOUT = "pollTimeout";
- public static final long POLL_TIMEOUT_DEFAULT = 1000L;
-
- public static final String CONVERTER = "converter";
-
- public static final String CONVERTER_TYPE = CONVERTER + ".type";
- public static final String CONVERTER_TYPE_DEFAULT = "DEFAULT";
-
- public static final String CONVERTER_CHARSET = CONVERTER + ".charset";
- public static final String CONVERTER_CHARSET_DEFAULT = "UTF-8";
-
- public static final String CREATE_DURABLE_SUBSCRIPTION = "createDurableSubscription";
- public static final boolean DEFAULT_CREATE_DURABLE_SUBSCRIPTION = false;
-
- public static final String DURABLE_SUBSCRIPTION_NAME = "durableSubscriptionName";
- public static final String DEFAULT_DURABLE_SUBSCRIPTION_NAME = "";
-
-}
diff --git a/flume-ng-sources/flume-jms-source/src/test/java/org/apache/flume/source/jms/JMSMessageConsumerTestBase.java b/flume-ng-sources/flume-jms-source/src/test/java/org/apache/flume/source/jms/JMSMessageConsumerTestBase.java
deleted file mode 100644
index aa964584a..000000000
--- a/flume-ng-sources/flume-jms-source/src/test/java/org/apache/flume/source/jms/JMSMessageConsumerTestBase.java
+++ /dev/null
@@ -1,151 +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.source.jms;
-
-import static org.junit.Assert.*;
-import static org.mockito.Matchers.*;
-import static org.mockito.Mockito.*;
-
-import java.util.Enumeration;
-import java.util.List;
-
-import javax.jms.Connection;
-import javax.jms.ConnectionFactory;
-import javax.jms.Destination;
-import javax.jms.MessageConsumer;
-import javax.jms.Queue;
-import javax.jms.Session;
-import javax.jms.TextMessage;
-import javax.jms.Topic;
-import javax.naming.InitialContext;
-
-import org.apache.flume.Context;
-import org.apache.flume.Event;
-import org.junit.After;
-import org.junit.Before;
-
-import com.google.common.base.Charsets;
-import com.google.common.base.Optional;
-
-public abstract class JMSMessageConsumerTestBase {
- static final String USERNAME = "userName";
- static final String PASSWORD = "password";
- static final String DESTINATION_NAME = "destinationName";
- static final String SELECTOR = "selector";
- static final String TEXT = "text";
- static final InitialContext WONT_USE = null;
-
- Context context;
- JMSMessageConsumer consumer;
- ConnectionFactory connectionFactory;
- String destinationName;
- JMSDestinationType destinationType;
- JMSDestinationLocator destinationLocator;
- String messageSelector;
- int batchSize;
- long pollTimeout;
- JMSMessageConverter converter;
- Optional<String> userName;
- Optional<String> password;
-
- Connection connection;
- Session session;
- Queue queue;
- Topic topic;
- MessageConsumer messageConsumer;
- TextMessage message;
- Event event;
-
- @Before
- public void setup() throws Exception {
- beforeSetup();
- connectionFactory = mock(ConnectionFactory.class);
- connection = mock(Connection.class);
- session = mock(Session.class);
- queue = mock(Queue.class);
- topic = mock(Topic.class);
- messageConsumer = mock(MessageConsumer.class);
- message = mock(TextMessage.class);
- when(message.getPropertyNames()).thenReturn(new Enumeration<Object>() {
- @Override
- public boolean hasMoreElements() {
- return false;
- }
- @Override
- public Object nextElement() {
- throw new UnsupportedOperationException();
- }
- });
- when(message.getText()).thenReturn(TEXT);
- when(connectionFactory.createConnection(USERNAME, PASSWORD)).thenReturn(connection);
- when(connection.createSession(true, Session.SESSION_TRANSACTED)).thenReturn(session);
- when(session.createQueue(destinationName)).thenReturn(queue);
- when(session.createConsumer(any(Destination.class), anyString())).thenReturn(messageConsumer);
- when(messageConsumer.receiveNoWait()).thenReturn(message);
- when(messageConsumer.receive(anyLong())).thenReturn(message);
- destinationName = DESTINATION_NAME;
- destinationType = JMSDestinationType.QUEUE;
- destinationLocator = JMSDestinationLocator.CDI;
- messageSelector = SELECTOR;
- batchSize = 10;
- pollTimeout = 500L;
- context = new Context();
- converter = new DefaultJMSMessageConverter.Builder().build(context);
- event = converter.convert(message).iterator().next();
- userName = Optional.of(USERNAME);
- password = Optional.of(PASSWORD);
- afterSetup();
- }
- void beforeSetup() throws Exception {
-
- }
- void afterSetup() throws Exception {
-
- }
- void beforeTearDown() throws Exception {
-
- }
- void afterTearDown() throws Exception {
-
- }
- void assertBodyIsExpected(List<Event> events) {
- for (Event event : events) {
- assertEquals(TEXT, new String(event.getBody(), Charsets.UTF_8));
- }
- }
-
- JMSMessageConsumer create(JMSDestinationType destinationType,
- JMSDestinationLocator destinationLocator, String destinationName) {
- return new JMSMessageConsumer(WONT_USE, connectionFactory, destinationName,
- destinationLocator, destinationType, messageSelector, batchSize,
- pollTimeout, converter, userName, password, Optional.<String>absent(), false, "");
- }
-
- JMSMessageConsumer create() {
- return create(this.destinationType, this.destinationLocator, this.destinationName);
- }
-
- @After
- public void tearDown() throws Exception {
- beforeTearDown();
- if (consumer != null) {
- consumer.close();
- }
- afterTearDown();
- }
-}
diff --git a/flume-ng-sources/flume-jms-source/src/test/java/org/apache/flume/source/jms/TestDefaultJMSMessageConverter.java b/flume-ng-sources/flume-jms-source/src/test/java/org/apache/flume/source/jms/TestDefaultJMSMessageConverter.java
deleted file mode 100644
index 6f0f1de0f..000000000
--- a/flume-ng-sources/flume-jms-source/src/test/java/org/apache/flume/source/jms/TestDefaultJMSMessageConverter.java
+++ /dev/null
@@ -1,189 +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.source.jms;
-
-import static org.junit.Assert.*;
-import static org.mockito.Matchers.*;
-import static org.mockito.Mockito.*;
-
-import java.io.ByteArrayOutputStream;
-import java.io.ObjectOutput;
-import java.io.ObjectOutputStream;
-import java.util.Collections;
-import java.util.Enumeration;
-import java.util.Iterator;
-import java.util.Map;
-
-import javax.jms.BytesMessage;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.ObjectMessage;
-import javax.jms.TextMessage;
-
-import org.apache.flume.Context;
-import org.apache.flume.Event;
-import org.junit.Before;
-import org.junit.Test;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
-
-import com.google.common.base.Charsets;
-import com.google.common.collect.Maps;
-
-public class TestDefaultJMSMessageConverter {
-
- private static final String TEXT = "text";
- private static final byte[] BYTES = TEXT.getBytes(Charsets.UTF_8);
- private Context context;
- private Message message;
- private Map<String, String> headers;
- private JMSMessageConverter converter;
-
- @Before
- public void setUp() throws Exception {
- headers = Maps.newHashMap();
- context = new Context();
- converter = new DefaultJMSMessageConverter.Builder().build(context);
- }
-
- void createTextMessage() throws Exception {
- TextMessage message = mock(TextMessage.class);
- when(message.getText()).thenReturn(TEXT);
- this.message = message;
- }
-
- void createNullTextMessage() throws Exception {
- TextMessage message = mock(TextMessage.class);
- when(message.getText()).thenReturn(null);
- this.message = message;
- }
-
- void createBytesMessage() throws Exception {
- BytesMessage message = mock(BytesMessage.class);
- when(message.getBodyLength()).thenReturn((long)BYTES.length);
- when(message.readBytes(any(byte[].class))).then(new Answer<Integer>() {
- @Override
- public Integer answer(InvocationOnMock invocation) throws Throwable {
- byte[] buffer = (byte[])invocation.getArguments()[0];
- if (buffer != null) {
- assertEquals(buffer.length, BYTES.length);
- System.arraycopy(BYTES, 0, buffer, 0, BYTES.length);
- }
- return BYTES.length;
- }
- });
- this.message = message;
- }
- void createObjectMessage() throws Exception {
- ObjectMessage message = mock(ObjectMessage.class);
- when(message.getObject()).thenReturn(TEXT);
- this.message = message;
- }
- void createHeaders() throws Exception {
- final Iterator<String> keys = headers.keySet().iterator();
- when(message.getPropertyNames()).thenReturn(new Enumeration<Object>() {
- @Override
- public boolean hasMoreElements() {
- return keys.hasNext();
- }
- @Override
- public Object nextElement() {
- return keys.next();
- }
- });
- when(message.getStringProperty(anyString())).then(new Answer<String>() {
- @Override
- public String answer(InvocationOnMock invocation) throws Throwable {
- return headers.get(invocation.getArguments()[0]);
- }
- });
- }
-
- @Test
- public void testTextMessage() throws Exception {
- createTextMessage();
- headers.put("key1", "value1");
- headers.put("key2", "value2");
- createHeaders();
- Event event = converter.convert(message).iterator().next();
- assertEquals(headers, event.getHeaders());
- assertEquals(TEXT, new String(event.getBody(), Charsets.UTF_8));
- }
-
- @Test
- public void testNullTextMessage() throws Exception {
- createNullTextMessage();
- headers.put("key1", "value1");
- headers.put("key2", "value2");
- createHeaders();
- Event event = converter.convert(message).iterator().next();
- assertEquals(headers, event.getHeaders());
- // In case of a null text message, the event's body will be empty due to
- // SimpleEvent's body not updated with a valid text message.
- assertEquals(event.getBody().length, 0);
- }
-
- @Test
- public void testBytesMessage() throws Exception {
- createBytesMessage();
- headers.put("key1", "value1");
- headers.put("key2", "value2");
- createHeaders();
- Event event = converter.convert(message).iterator().next();
- assertEquals(headers, event.getHeaders());
- assertArrayEquals(BYTES, event.getBody());
- }
- @Test(expected = JMSException.class)
- public void testBytesMessageTooLarge() throws Exception {
- createBytesMessage();
- when(((BytesMessage)message).getBodyLength()).thenReturn(Long.MAX_VALUE);
- createHeaders();
- converter.convert(message);
- }
- @Test(expected = JMSException.class)
- public void testBytesMessagePartialReturn() throws Exception {
- createBytesMessage();
- when(((BytesMessage)message).readBytes(any(byte[].class)))
- .thenReturn(BYTES.length + 1);
- createHeaders();
- converter.convert(message);
- }
-
- @Test
- public void testObjectMessage() throws Exception {
- createObjectMessage();
- headers.put("key1", "value1");
- headers.put("key2", "value2");
- createHeaders();
- Event event = converter.convert(message).iterator().next();
- assertEquals(headers, event.getHeaders());
- ByteArrayOutputStream bos = new ByteArrayOutputStream();
- ObjectOutput out = new ObjectOutputStream(bos);
- out.writeObject(TEXT);
- assertArrayEquals(bos.toByteArray(), event.getBody());
- }
-
- @Test
- public void testNoHeaders() throws Exception {
- createTextMessage();
- createHeaders();
- Event event = converter.convert(message).iterator().next();
- assertEquals(Collections.EMPTY_MAP, event.getHeaders());
- assertEquals(TEXT, new String(event.getBody(), Charsets.UTF_8));
- }
-}
diff --git a/flume-ng-sources/flume-jms-source/src/test/java/org/apache/flume/source/jms/TestIntegrationActiveMQ.java b/flume-ng-sources/flume-jms-source/src/test/java/org/apache/flume/source/jms/TestIntegrationActiveMQ.java
deleted file mode 100644
index 10320d749..000000000
--- a/flume-ng-sources/flume-jms-source/src/test/java/org/apache/flume/source/jms/TestIntegrationActiveMQ.java
+++ /dev/null
@@ -1,334 +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.source.jms;
-
-import static org.mockito.Matchers.*;
-import static org.mockito.Mockito.*;
-
-import java.io.File;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.List;
-import java.util.UUID;
-
-import javax.jms.Connection;
-import javax.jms.ConnectionFactory;
-import javax.jms.Destination;
-import javax.jms.MessageProducer;
-import javax.jms.Session;
-import javax.jms.TextMessage;
-
-import junit.framework.Assert;
-
-import org.apache.activemq.ActiveMQConnectionFactory;
-import org.apache.activemq.broker.BrokerPlugin;
-import org.apache.activemq.broker.BrokerService;
-import org.apache.activemq.security.AuthenticationUser;
-import org.apache.activemq.security.SimpleAuthenticationPlugin;
-import org.apache.commons.io.FileUtils;
-import org.apache.flume.Context;
-import org.apache.flume.Event;
-import org.apache.flume.PollableSource.Status;
-import org.apache.flume.channel.ChannelProcessor;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
-
-import com.google.common.base.Charsets;
-import com.google.common.collect.Lists;
-import com.google.common.io.Files;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-@RunWith(Parameterized.class)
-public class TestIntegrationActiveMQ {
-
- private static final Logger LOGGER = LoggerFactory.getLogger(TestIntegrationActiveMQ.class);
-
- private static final String INITIAL_CONTEXT_FACTORY =
- "org.apache.activemq.jndi.ActiveMQInitialContextFactory";
- public static final String BROKER_BIND_URL = "tcp://localhost:61516";
- private static final String DESTINATION_NAME = "test";
- // specific for dynamic queues on ActiveMq
- public static final String JNDI_PREFIX = "dynamicQueues/";
-
- private enum TestMode {
- WITH_AUTHENTICATION,
- WITHOUT_AUTHENTICATION
- }
-
- private File baseDir;
- private File tmpDir;
- private File dataDir;
-
- private BrokerService broker;
- private Context context;
- private JMSSource source;
- private List<Event> events;
-
- private final String jmsUserName;
- private final String jmsPassword;
-
- public TestIntegrationActiveMQ(TestMode testMode) {
- System.setProperty(JMSSource.JNDI_ALLOWED_PROTOCOLS, "tcp");
- LOGGER.info("Testing with test mode {}", testMode);
-
- switch (testMode) {
- case WITH_AUTHENTICATION:
- jmsUserName = "user";
- jmsPassword = "pass";
- break;
- case WITHOUT_AUTHENTICATION:
- jmsUserName = null;
- jmsPassword = null;
- break;
- default:
- throw new IllegalArgumentException("Unhandled test mode: " + testMode);
- }
- }
-
- @Parameterized.Parameters
- public static Collection<Object[]> parameters() {
- return Arrays.asList(new Object[][]{
- {TestMode.WITH_AUTHENTICATION},
- {TestMode.WITHOUT_AUTHENTICATION}
- });
- }
-
- @SuppressWarnings("unchecked")
- @Before
- public void setup() throws Exception {
- baseDir = Files.createTempDir();
- tmpDir = new File(baseDir, "tmp");
- dataDir = new File(baseDir, "data");
- Assert.assertTrue(tmpDir.mkdir());
-
- broker = new BrokerService();
- broker.addConnector(BROKER_BIND_URL);
- broker.setTmpDataDirectory(tmpDir);
- broker.setDataDirectoryFile(dataDir);
-
- context = new Context();
- context.put(JMSSourceConfiguration.INITIAL_CONTEXT_FACTORY, INITIAL_CONTEXT_FACTORY);
- context.put(JMSSourceConfiguration.PROVIDER_URL, BROKER_BIND_URL);
- context.put(JMSSourceConfiguration.DESTINATION_NAME, DESTINATION_NAME);
-
- if (jmsUserName != null) {
- File passwordFile = new File(baseDir, "password");
- Files.write(jmsPassword.getBytes(Charsets.UTF_8), passwordFile);
-
- AuthenticationUser jmsUser = new AuthenticationUser(jmsUserName, jmsPassword, "");
- List<AuthenticationUser> users = Collections.singletonList(jmsUser);
- SimpleAuthenticationPlugin authentication = new SimpleAuthenticationPlugin(users);
- broker.setPlugins(new BrokerPlugin[]{authentication});
-
- context.put(JMSSourceConfiguration.USERNAME, jmsUserName);
- context.put(JMSSourceConfiguration.PASSWORD_FILE, passwordFile.getAbsolutePath());
- }
-
- broker.start();
-
- events = Lists.newArrayList();
- source = new JMSSource();
- source.setName("JMSSource-" + UUID.randomUUID());
- ChannelProcessor channelProcessor = mock(ChannelProcessor.class);
- doAnswer(new Answer<Void>() {
- @Override
- public Void answer(InvocationOnMock invocation) throws Throwable {
- events.addAll((List<Event>)invocation.getArguments()[0]);
- return null;
- }
- }).when(channelProcessor).processEventBatch(any(List.class));
- source.setChannelProcessor(channelProcessor);
- }
-
- @After
- public void tearDown() throws Exception {
- if (source != null) {
- source.stop();
- }
- if (broker != null) {
- broker.stop();
- }
- FileUtils.deleteDirectory(baseDir);
- }
-
- private void putQueue(List<String> events) throws Exception {
- ConnectionFactory factory = new ActiveMQConnectionFactory(jmsUserName, jmsPassword,
- BROKER_BIND_URL);
- Connection connection = factory.createConnection();
- connection.start();
-
- Session session = connection.createSession(true,
- Session.AUTO_ACKNOWLEDGE);
- Destination destination = session.createQueue(DESTINATION_NAME);
- MessageProducer producer = session.createProducer(destination);
-
- for (String event : events) {
- TextMessage message = session.createTextMessage();
- message.setText(event);
- producer.send(message);
- }
- session.commit();
- session.close();
- connection.close();
- }
-
- private void putTopic(List<String> events) throws Exception {
- ConnectionFactory factory = new ActiveMQConnectionFactory(jmsUserName, jmsPassword,
- BROKER_BIND_URL);
- Connection connection = factory.createConnection();
- connection.start();
-
- Session session = connection.createSession(true,
- Session.AUTO_ACKNOWLEDGE);
- Destination destination = session.createTopic(DESTINATION_NAME);
- MessageProducer producer = session.createProducer(destination);
-
- for (String event : events) {
- TextMessage message = session.createTextMessage();
- message.setText(event);
- producer.send(message);
- }
- session.commit();
- session.close();
- connection.close();
- }
-
- @Test
- public void testQueueLocatedWithJndi() throws Exception {
- context.put(JMSSourceConfiguration.DESTINATION_NAME,
- JNDI_PREFIX + DESTINATION_NAME);
- context.put(JMSSourceConfiguration.DESTINATION_LOCATOR,
- JMSDestinationLocator.JNDI.name());
- testQueue();
- }
-
- @Test
- public void testQueue() throws Exception {
- context.put(JMSSourceConfiguration.DESTINATION_TYPE,
- JMSSourceConfiguration.DESTINATION_TYPE_QUEUE);
- source.configure(context);
- source.start();
- Thread.sleep(500L);
-
- List<String> expected = Lists.newArrayList();
- for (int i = 0; i < 10; i++) {
- expected.add(String.valueOf(i));
- }
- putQueue(expected);
-
- Thread.sleep(500L);
-
- Assert.assertEquals(Status.READY, source.process());
- Assert.assertEquals(Status.BACKOFF, source.process());
- Assert.assertEquals(expected.size(), events.size());
- List<String> actual = Lists.newArrayList();
- for (Event event : events) {
- actual.add(new String(event.getBody(), Charsets.UTF_8));
- }
- Collections.sort(expected);
- Collections.sort(actual);
- Assert.assertEquals(expected, actual);
- }
-
- @Test
- public void testTopic() throws Exception {
- context.put(JMSSourceConfiguration.DESTINATION_TYPE,
- JMSSourceConfiguration.DESTINATION_TYPE_TOPIC);
- source.configure(context);
- source.start();
- Thread.sleep(500L);
-
- List<String> expected = Lists.newArrayList();
- for (int i = 0; i < 10; i++) {
- expected.add(String.valueOf(i));
- }
- putTopic(expected);
-
- Thread.sleep(500L);
-
- Assert.assertEquals(Status.READY, source.process());
- Assert.assertEquals(Status.BACKOFF, source.process());
- Assert.assertEquals(expected.size(), events.size());
- List<String> actual = Lists.newArrayList();
- for (Event event : events) {
- actual.add(new String(event.getBody(), Charsets.UTF_8));
- }
- Collections.sort(expected);
- Collections.sort(actual);
- Assert.assertEquals(expected, actual);
- }
-
- @Test
- public void testDurableSubscription() throws Exception {
- context.put(JMSSourceConfiguration.DESTINATION_TYPE,
- JMSSourceConfiguration.DESTINATION_TYPE_TOPIC);
- context.put(JMSSourceConfiguration.CLIENT_ID, "FLUME");
- context.put(JMSSourceConfiguration.DURABLE_SUBSCRIPTION_NAME, "SOURCE1");
- context.put(JMSSourceConfiguration.CREATE_DURABLE_SUBSCRIPTION, "true");
- context.put(JMSSourceConfiguration.BATCH_SIZE, "10");
- source.configure(context);
- source.start();
- Thread.sleep(5000L);
- List<String> expected = Lists.newArrayList();
- List<String> input = Lists.newArrayList();
- for (int i = 0; i < 10; i++) {
- input.add("before " + String.valueOf(i));
- }
- expected.addAll(input);
- putTopic(input);
-
- Thread.sleep(500L);
- Assert.assertEquals(Status.READY, source.process());
- Assert.assertEquals(Status.BACKOFF, source.process());
- source.stop();
- Thread.sleep(500L);
- input = Lists.newArrayList();
- for (int i = 0; i < 10; i++) {
- input.add("during " + String.valueOf(i));
- }
- expected.addAll(input);
- putTopic(input);
- source.start();
- Thread.sleep(500L);
- input = Lists.newArrayList();
- for (int i = 0; i < 10; i++) {
- input.add("after " + String.valueOf(i));
- }
- expected.addAll(input);
- putTopic(input);
-
- Assert.assertEquals(Status.READY, source.process());
- Assert.assertEquals(Status.READY, source.process());
- Assert.assertEquals(Status.BACKOFF, source.process());
- Assert.assertEquals(expected.size(), events.size());
- List<String> actual = Lists.newArrayList();
- for (Event event : events) {
- actual.add(new String(event.getBody(), Charsets.UTF_8));
- }
- Collections.sort(expected);
- Collections.sort(actual);
- Assert.assertEquals(expected, actual);
- }
-}
\ No newline at end of file
diff --git a/flume-ng-sources/flume-jms-source/src/test/java/org/apache/flume/source/jms/TestJMSMessageConsumer.java b/flume-ng-sources/flume-jms-source/src/test/java/org/apache/flume/source/jms/TestJMSMessageConsumer.java
deleted file mode 100644
index 636ffc322..000000000
--- a/flume-ng-sources/flume-jms-source/src/test/java/org/apache/flume/source/jms/TestJMSMessageConsumer.java
+++ /dev/null
@@ -1,280 +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.source.jms;
-
-import static org.junit.Assert.*;
-import static org.mockito.Matchers.*;
-import static org.mockito.Mockito.*;
-
-import java.util.List;
-
-import javax.jms.Destination;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.Session;
-import javax.jms.Topic;
-import javax.jms.TopicSubscriber;
-
-import org.apache.flume.Event;
-import org.apache.flume.FlumeException;
-import org.junit.Test;
-
-import com.google.common.base.Optional;
-
-public class TestJMSMessageConsumer extends JMSMessageConsumerTestBase {
-
- @Test(expected = FlumeException.class)
- public void testCreateConnectionFails() throws Exception {
- when(connectionFactory.createConnection(USERNAME, PASSWORD))
- .thenThrow(new JMSException(""));
- create();
- }
-
- @Test
- public void testCreateSessionFails() throws Exception {
- when(connection.createSession(true, Session.SESSION_TRANSACTED))
- .thenThrow(new JMSException(""));
- try {
- create();
- fail("Expected exception: org.apache.flume.FlumeException");
- } catch (FlumeException e) {
- verify(connection).close();
- }
- }
- @Test
- public void testCreateQueueFails() throws Exception {
- when(session.createQueue(destinationName))
- .thenThrow(new JMSException(""));
- try {
- create();
- fail("Expected exception: org.apache.flume.FlumeException");
- } catch (FlumeException e) {
- verify(session).close();
- verify(connection).close();
- }
- }
- @Test
- public void testCreateTopicFails() throws Exception {
- destinationType = JMSDestinationType.TOPIC;
- when(session.createTopic(destinationName))
- .thenThrow(new JMSException(""));
- try {
- create();
- fail("Expected exception: org.apache.flume.FlumeException");
- } catch (FlumeException e) {
- verify(session).close();
- verify(connection).close();
- }
- }
- @Test
- public void testCreateConsumerFails() throws Exception {
- when(session.createConsumer(any(Destination.class), anyString()))
- .thenThrow(new JMSException(""));
- try {
- create();
- fail("Expected exception: org.apache.flume.FlumeException");
- } catch (FlumeException e) {
- verify(session).close();
- verify(connection).close();
- }
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testInvalidDestination() throws Exception {
- create(null, JMSDestinationLocator.JNDI, "ldap://localhost:389/test");
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testInvalidBatchSizeZero() throws Exception {
- batchSize = 0;
- create();
- }
- @Test(expected = IllegalArgumentException.class)
- public void testInvalidPollTime() throws Exception {
- pollTimeout = -1L;
- create();
- }
- @Test(expected = IllegalArgumentException.class)
- public void testInvalidBatchSizeNegative() throws Exception {
- batchSize = -1;
- create();
- }
-
- @Test
- public void testQueue() throws Exception {
- destinationType = JMSDestinationType.QUEUE;
- when(session.createQueue(destinationName)).thenReturn(queue);
- consumer = create();
- List<Event> events = consumer.take();
- assertEquals(batchSize, events.size());
- assertBodyIsExpected(events);
- verify(session, never()).createTopic(anyString());
- }
- @Test
- public void testTopic() throws Exception {
- destinationType = JMSDestinationType.TOPIC;
- when(session.createTopic(destinationName)).thenReturn(topic);
- consumer = create();
- List<Event> events = consumer.take();
- assertEquals(batchSize, events.size());
- assertBodyIsExpected(events);
- verify(session, never()).createQueue(anyString());
- }
- @Test
- public void testUserPass() throws Exception {
- consumer = create();
- List<Event> events = consumer.take();
- assertEquals(batchSize, events.size());
- assertBodyIsExpected(events);
- }
- @Test
- public void testNoUserPass() throws Exception {
- userName = Optional.absent();
- when(connectionFactory.createConnection(USERNAME, PASSWORD)).thenThrow(new AssertionError());
- when(connectionFactory.createConnection()).thenReturn(connection);
- consumer = create();
- List<Event> events = consumer.take();
- assertEquals(batchSize, events.size());
- assertBodyIsExpected(events);
- }
- @Test
- public void testNoEvents() throws Exception {
- when(messageConsumer.receive(anyLong())).thenReturn(null);
- consumer = create();
- List<Event> events = consumer.take();
- assertEquals(0, events.size());
- verify(messageConsumer, times(1)).receive(anyLong());
- verifyNoMoreInteractions(messageConsumer);
- }
- @Test
- public void testSingleEvent() throws Exception {
- when(messageConsumer.receiveNoWait()).thenReturn(null);
- consumer = create();
- List<Event> events = consumer.take();
- assertEquals(1, events.size());
- assertBodyIsExpected(events);
- }
- @Test
- public void testPartialBatch() throws Exception {
- when(messageConsumer.receiveNoWait()).thenReturn(message, (Message)null);
- consumer = create();
- List<Event> events = consumer.take();
- assertEquals(2, events.size());
- assertBodyIsExpected(events);
- }
- @Test
- public void testCommit() throws Exception {
- consumer = create();
- consumer.commit();
- verify(session, times(1)).commit();
- }
- @Test
- public void testRollback() throws Exception {
- consumer = create();
- consumer.rollback();
- verify(session, times(1)).rollback();
- }
- @Test
- public void testClose() throws Exception {
- doThrow(new JMSException("")).when(session).close();
- consumer = create();
- consumer.close();
- verify(session, times(1)).close();
- verify(connection, times(1)).close();
- }
-
- @Test
- public void testCreateDurableSubscription() throws Exception {
- String name = "SUBSCRIPTION_NAME";
- String clientID = "CLIENT_ID";
- TopicSubscriber mockTopicSubscriber = mock(TopicSubscriber.class);
- when(session.createDurableSubscriber(any(Topic.class), anyString(), anyString(), anyBoolean()))
- .thenReturn(mockTopicSubscriber );
- when(session.createTopic(destinationName)).thenReturn(topic);
- new JMSMessageConsumer(WONT_USE, connectionFactory, destinationName, destinationLocator,
- JMSDestinationType.TOPIC, messageSelector, batchSize, pollTimeout, converter, userName,
- password, Optional.of(clientID), true, name);
- verify(connection, times(1)).setClientID(clientID);
- verify(session, times(1)).createDurableSubscriber(topic, name, messageSelector, true);
- }
-
- @Test(expected = JMSException.class)
- public void testTakeFailsDueToJMSExceptionFromReceive() throws JMSException {
- when(messageConsumer.receive(anyLong())).thenThrow(new JMSException(""));
- consumer = create();
-
- consumer.take();
- }
-
- @Test(expected = JMSException.class)
- public void testTakeFailsDueToRuntimeExceptionFromReceive() throws JMSException {
- when(messageConsumer.receive(anyLong())).thenThrow(new RuntimeException());
- consumer = create();
-
- consumer.take();
- }
-
- @Test(expected = JMSException.class)
- public void testTakeFailsDueToJMSExceptionFromReceiveNoWait() throws JMSException {
- when(messageConsumer.receiveNoWait()).thenThrow(new JMSException(""));
- consumer = create();
-
- consumer.take();
- }
-
- @Test(expected = JMSException.class)
- public void testTakeFailsDueToRuntimeExceptionFromReceiveNoWait() throws JMSException {
- when(messageConsumer.receiveNoWait()).thenThrow(new RuntimeException());
- consumer = create();
-
- consumer.take();
- }
-
- @Test
- public void testCommitFailsDueToJMSException() throws JMSException {
- doThrow(new JMSException("")).when(session).commit();
- consumer = create();
-
- consumer.commit();
- }
-
- @Test
- public void testCommitFailsDueToRuntimeException() throws JMSException {
- doThrow(new RuntimeException()).when(session).commit();
- consumer = create();
-
- consumer.commit();
- }
-
- @Test
- public void testRollbackFailsDueToJMSException() throws JMSException {
- doThrow(new JMSException("")).when(session).rollback();
- consumer = create();
-
- consumer.rollback();
- }
-
- @Test
- public void testRollbackFailsDueToRuntimeException() throws JMSException {
- doThrow(new RuntimeException()).when(session).rollback();
- consumer = create();
-
- consumer.rollback();
- }
-
-}
diff --git a/flume-ng-sources/flume-jms-source/src/test/java/org/apache/flume/source/jms/TestJMSSource.java b/flume-ng-sources/flume-jms-source/src/test/java/org/apache/flume/source/jms/TestJMSSource.java
deleted file mode 100644
index ea17284f2..000000000
--- a/flume-ng-sources/flume-jms-source/src/test/java/org/apache/flume/source/jms/TestJMSSource.java
+++ /dev/null
@@ -1,384 +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.source.jms;
-
-import static org.mockito.Matchers.*;
-import static org.mockito.Mockito.*;
-
-import java.io.File;
-import java.util.List;
-import java.util.Properties;
-import java.util.UUID;
-
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.naming.InitialContext;
-import javax.naming.NamingException;
-
-import junit.framework.Assert;
-
-import org.apache.commons.io.FileUtils;
-import org.apache.flume.exception.ChannelException;
-import org.apache.flume.Context;
-import org.apache.flume.Event;
-import org.apache.flume.EventDeliveryException;
-import org.apache.flume.FlumeException;
-import org.apache.flume.PollableSource.Status;
-import org.apache.flume.channel.ChannelProcessor;
-import org.apache.flume.conf.Configurable;
-import org.apache.flume.instrumentation.SourceCounter;
-import org.junit.Test;
-import org.mockito.internal.util.reflection.Whitebox;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
-
-import com.google.common.collect.Lists;
-import com.google.common.io.Files;
-
-public class TestJMSSource extends JMSMessageConsumerTestBase {
-
- private JMSSource source;
- private Context context;
- private InitialContext initialContext;
- private ChannelProcessor channelProcessor;
- private List<Event> events;
- private InitialContextFactory contextFactory;
- private File baseDir;
- private File passwordFile;
-
- @SuppressWarnings("unchecked")
- @Override
- void afterSetup() throws Exception {
- System.setProperty(JMSSource.JNDI_ALLOWED_PROTOCOLS, "dummy");
- baseDir = Files.createTempDir();
- passwordFile = new File(baseDir, "password");
- Assert.assertTrue(passwordFile.createNewFile());
- initialContext = mock(InitialContext.class);
- channelProcessor = mock(ChannelProcessor.class);
- events = Lists.newArrayList();
- doAnswer(new Answer<Void>() {
- @Override
- public Void answer(InvocationOnMock invocation) throws Throwable {
- events.addAll((List<Event>)invocation.getArguments()[0]);
- return null;
- }
- }).when(channelProcessor).processEventBatch(any(List.class));
- consumer = spy(create());
- when(initialContext.lookup(anyString())).thenReturn(connectionFactory);
- contextFactory = mock(InitialContextFactory.class);
- when(contextFactory.create(any(Properties.class))).thenReturn(initialContext);
- source = spy(new JMSSource(contextFactory));
- doReturn(consumer).when(source).createConsumer();
- source.setName("JMSSource-" + UUID.randomUUID());
- source.setChannelProcessor(channelProcessor);
- context = new Context();
- context.put(JMSSourceConfiguration.BATCH_SIZE, String.valueOf(batchSize));
- context.put(JMSSourceConfiguration.DESTINATION_NAME, "INBOUND");
- context.put(JMSSourceConfiguration.DESTINATION_TYPE,
- JMSSourceConfiguration.DESTINATION_TYPE_QUEUE);
- context.put(JMSSourceConfiguration.PROVIDER_URL, "dummy:1414");
- context.put(JMSSourceConfiguration.INITIAL_CONTEXT_FACTORY, "ldap://dummy:389");
- }
-
- @Override
- void afterTearDown() throws Exception {
- FileUtils.deleteDirectory(baseDir);
- }
-
- @Test
- public void testStop() throws Exception {
- source.configure(context);
- source.start();
- source.stop();
- verify(consumer).close();
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testConfigureWithoutInitialContextFactory() throws Exception {
- context.put(JMSSourceConfiguration.INITIAL_CONTEXT_FACTORY, "");
- source.configure(context);
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testConfigureWithoutProviderURL() throws Exception {
- context.put(JMSSourceConfiguration.PROVIDER_URL, "");
- source.configure(context);
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testConfigureWithoutDestinationName() throws Exception {
- context.put(JMSSourceConfiguration.DESTINATION_NAME, "");
- source.configure(context);
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testConfigureWithConnectionFactory() throws Exception {
- context.put(JMSSourceConfiguration.CONNECTION_FACTORY,
- "ldap://localhost:319/connectionFactory");
- source.configure(context);
- }
-
- @Test(expected = FlumeException.class)
- public void testConfigureWithBadDestinationType() throws Exception {
- context.put(JMSSourceConfiguration.DESTINATION_TYPE, "DUMMY");
- source.configure(context);
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testConfigureWithEmptyDestinationType() throws Exception {
- context.put(JMSSourceConfiguration.DESTINATION_TYPE, "");
- source.configure(context);
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testConfigureWithLdapProvider() throws Exception {
- context.put(JMSSourceConfiguration.PROVIDER_URL, "ldap://localhost:389/test");
- source.configure(context);
- }
-
-
- @Test
- public void testStartConsumerCreateThrowsException() throws Exception {
- doThrow(new RuntimeException("Expected")).when(source).createConsumer();
- source.configure(context);
- source.start();
- try {
- source.process();
- Assert.fail();
- } catch (FlumeException expected) {
-
- }
- }
-
- @Test(expected = FlumeException.class)
- public void testConfigureWithContextLookupThrowsException() throws Exception {
- when(initialContext.lookup(anyString())).thenThrow(new NamingException());
- source.configure(context);
- }
-
- @Test(expected = FlumeException.class)
- public void testConfigureWithContextCreateThrowsException() throws Exception {
- when(contextFactory.create(any(Properties.class)))
- .thenThrow(new NamingException());
- source.configure(context);
- }
-
- @Test(expected = IllegalArgumentException.class)
- public void testConfigureWithInvalidBatchSize() throws Exception {
- context.put(JMSSourceConfiguration.BATCH_SIZE, "0");
- source.configure(context);
- }
-
- @Test(expected = FlumeException.class)
- public void testConfigureWithInvalidPasswordFile() throws Exception {
- context.put(JMSSourceConfiguration.PASSWORD_FILE,
- "/dev/does/not/exist/nor/will/ever/exist");
- source.configure(context);
- }
-
- @Test
- public void testConfigureWithUserNameButNoPasswordFile() throws Exception {
- context.put(JMSSourceConfiguration.USERNAME, "dummy");
- source.configure(context);
- source.start();
- Assert.assertEquals(Status.READY, source.process());
- Assert.assertEquals(batchSize, events.size());
- assertBodyIsExpected(events);
- }
-
- @Test
- public void testConfigureWithUserNameAndPasswordFile() throws Exception {
- context.put(JMSSourceConfiguration.USERNAME, "dummy");
- context.put(JMSSourceConfiguration.PASSWORD_FILE,
- passwordFile.getAbsolutePath());
- source.configure(context);
- source.start();
- Assert.assertEquals(Status.READY, source.process());
- Assert.assertEquals(batchSize, events.size());
- assertBodyIsExpected(events);
- }
-
- @Test(expected = FlumeException.class)
- public void testConfigureWithInvalidConverterClass() throws Exception {
- context.put(JMSSourceConfiguration.CONVERTER_TYPE, "not a valid classname");
- source.configure(context);
- }
-
- @Test
- public void testProcessNoStart() throws Exception {
- try {
- source.process();
- Assert.fail();
- } catch (EventDeliveryException expected) {
-
- }
- }
-
- @Test
- public void testNonDefaultConverter() throws Exception {
- // tests that a classname can be specified
- context.put(JMSSourceConfiguration.CONVERTER_TYPE,
- DefaultJMSMessageConverter.Builder.class.getName());
- source.configure(context);
- source.start();
- Assert.assertEquals(Status.READY, source.process());
- Assert.assertEquals(batchSize, events.size());
- assertBodyIsExpected(events);
- verify(consumer).commit();
- }
-
- public static class NonBuilderNonConfigurableConverter implements JMSMessageConverter {
- @Override
- public List<Event> convert(Message message) throws JMSException {
- throw new UnsupportedOperationException();
- }
- }
-
- public static class NonBuilderConfigurableConverter implements JMSMessageConverter, Configurable {
- @Override
- public List<Event> convert(Message message) throws JMSException {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void configure(Context context) {
-
- }
- }
-
- @Test
- public void testNonBuilderConfigurableConverter() throws Exception {
- // tests that a non builder by configurable converter works
- context.put(JMSSourceConfiguration.CONVERTER_TYPE,
- NonBuilderConfigurableConverter.class.getName());
- source.configure(context);
- source.start();
- Assert.assertEquals(Status.READY, source.process());
- Assert.assertEquals(batchSize, events.size());
- assertBodyIsExpected(events);
- verify(consumer).commit();
- }
-
- @Test
- public void testNonBuilderNonConfigurableConverter() throws Exception {
- // tests that a non builder non configurable converter
- context.put(JMSSourceConfiguration.CONVERTER_TYPE,
- NonBuilderNonConfigurableConverter.class.getName());
- source.configure(context);
- source.start();
- Assert.assertEquals(Status.READY, source.process());
- Assert.assertEquals(batchSize, events.size());
- assertBodyIsExpected(events);
- verify(consumer).commit();
- }
-
- @Test
- public void testProcessFullBatch() throws Exception {
- source.configure(context);
- source.start();
- Assert.assertEquals(Status.READY, source.process());
- Assert.assertEquals(batchSize, events.size());
- assertBodyIsExpected(events);
- verify(consumer).commit();
- }
-
- @Test
- public void testProcessNoEvents() throws Exception {
- when(messageConsumer.receive(anyLong())).thenReturn(null);
- source.configure(context);
- source.start();
- Assert.assertEquals(Status.BACKOFF, source.process());
- Assert.assertEquals(0, events.size());
- verify(consumer).commit();
- }
-
- @Test
- public void testProcessPartialBatch() throws Exception {
- when(messageConsumer.receiveNoWait()).thenReturn(message, (Message)null);
- source.configure(context);
- source.start();
- Assert.assertEquals(Status.READY, source.process());
- Assert.assertEquals(2, events.size());
- assertBodyIsExpected(events);
- verify(consumer).commit();
- }
-
- @SuppressWarnings("unchecked")
- @Test
- public void testProcessChannelProcessorThrowsChannelException() throws Exception {
- doThrow(new ChannelException("dummy"))
- .when(channelProcessor).processEventBatch(any(List.class));
- source.configure(context);
- source.start();
- Assert.assertEquals(Status.BACKOFF, source.process());
- verify(consumer).rollback();
- }
-
- @SuppressWarnings("unchecked")
- @Test
- public void testProcessChannelProcessorThrowsError() throws Exception {
- doThrow(new Error())
- .when(channelProcessor).processEventBatch(any(List.class));
- source.configure(context);
- source.start();
- try {
- source.process();
- Assert.fail();
- } catch (Error ignores) {
-
- }
- verify(consumer).rollback();
- }
-
- @Test
- public void testProcessReconnect() throws Exception {
- source.configure(context);
- source.start();
- when(consumer.take()).thenThrow(new JMSException("dummy"));
- int attempts = JMSSourceConfiguration.ERROR_THRESHOLD_DEFAULT;
- for (int i = 0; i < attempts; i++) {
- Assert.assertEquals(Status.BACKOFF, source.process());
- }
- Assert.assertEquals(Status.BACKOFF, source.process());
- SourceCounter sc = (SourceCounter) Whitebox.getInternalState(source, "sourceCounter");
- Assert.assertEquals(1, sc.getEventReadFail());
- verify(consumer, times(attempts + 1)).rollback();
- verify(consumer, times(1)).close();
- }
-
- @Test
- public void testErrorCounterEventReadFail() throws Exception {
- source.configure(context);
- source.start();
- when(consumer.take()).thenThrow(new RuntimeException("dummy"));
- source.process();
- SourceCounter sc = (SourceCounter) Whitebox.getInternalState(source, "sourceCounter");
- Assert.assertEquals(1, sc.getEventReadFail());
- }
-
- @Test
- public void testErrorCounterChannelWriteFail() throws Exception {
- source.configure(context);
- source.start();
- when(source.getChannelProcessor()).thenThrow(new ChannelException("dummy"));
- source.process();
- SourceCounter sc = (SourceCounter) Whitebox.getInternalState(source, "sourceCounter");
- Assert.assertEquals(1, sc.getChannelWriteFail());
- }
-
-}
\ No newline at end of file
diff --git a/flume-ng-sources/flume-jms-source/src/test/java/org/apache/flume/source/jms/TestJMSSourceCreation.java b/flume-ng-sources/flume-jms-source/src/test/java/org/apache/flume/source/jms/TestJMSSourceCreation.java
deleted file mode 100644
index bba1a977e..000000000
--- a/flume-ng-sources/flume-jms-source/src/test/java/org/apache/flume/source/jms/TestJMSSourceCreation.java
+++ /dev/null
@@ -1,49 +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.source.jms;
-
-import org.apache.flume.FlumeException;
-import org.apache.flume.Source;
-import org.apache.flume.SourceFactory;
-import org.apache.flume.source.DefaultSourceFactory;
-import org.junit.Assert;
-import org.junit.Before;
-import org.junit.Test;
-
-public class TestJMSSourceCreation {
-
- private SourceFactory sourceFactory;
-
- @Before
- public void setUp() {
- sourceFactory = new DefaultSourceFactory();
- }
-
- private void verifySourceCreation(String name, String type,
- Class<?> typeClass) throws FlumeException {
- Source src = sourceFactory.create(name, type);
- Assert.assertNotNull(src);
- Assert.assertTrue(typeClass.isInstance(src));
- }
-
- @Test
- public void testJMSSourceCreation() {
- verifySourceCreation("jms-src", "jms", JMSSource.class);
- }
-}
diff --git a/flume-ng-sources/flume-scribe-source/pom.xml b/flume-ng-sources/flume-scribe-source/pom.xml
deleted file mode 100644
index 83405dc65..000000000
--- a/flume-ng-sources/flume-scribe-source/pom.xml
+++ /dev/null
@@ -1,184 +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-sources</artifactId>
- <groupId>org.apache.flume</groupId>
- <version>2.0.0-SNAPSHOT</version>
- </parent>
-
- <groupId>org.apache.flume.flume-ng-sources</groupId>
- <artifactId>flume-scribe-source</artifactId>
- <name>Flume Scribe Source</name>
-
- <properties>
- <!-- TODO fix spotbugs/pmd violations -->
- <spotbugs.maxAllowedViolations>19</spotbugs.maxAllowedViolations>
- <pmd.maxAllowedViolations>76</pmd.maxAllowedViolations>
- <module.name>org.apache.flume.source.scribe</module.name>
- </properties>
-
- <profiles>
- <profile>
- <id>compileThrift</id>
- <activation>
- <activeByDefault>false</activeByDefault>
- </activation>
- <build>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-antrun-plugin</artifactId>
- <executions>
- <execution>
- <id>compileThrift</id>
- <phase>generate-sources</phase>
- <goals>
- <goal>run</goal>
- </goals>
- <configuration>
- <target>
- <echo file="target/compile-thrift.sh">
- LICENSE=src/main/thrift/aslv2
- THRIFT_DIR=src/main/thrift
- JAVA_DIR=target/generated-sources/thrift
- mkdir -p $JAVA_DIR 2> /dev/null
- JSTATUS=$?
- if [ $JSTATUS -ne 0 ] ; then
- echo "Could not create $JAVA_DIR. Will not generate thrift files."
- exit $JSTATUS
- fi
- for THRIFT_FILE in `ls $THRIFT_DIR/*.thrift 2> /dev/null`
- do
- thrift --gen java:hashcode -o $JAVA_DIR $THRIFT_FILE
- done
- SRC_DIR=$JAVA_DIR/gen-java/org/apache/flume/source/scribe
- DEST_DIR=src/main/java/org/apache/flume/source/scribe
- if [ ! -d $DEST_DIR ] ; then
- mkdir $DEST_DIR 2> /dev/null
- STATUS=$?
- if [ $STATUS -ne 0 ] ; then
- echo "Could not create $DEST_DIR. Will not generate thrift files."
- exit $STATUS
- fi
- fi
- for JAVA_FILE in `ls $SRC_DIR/*.java 2> /dev/null`
- do
- echo $JAVA_FILE
- cat $LICENSE > $JAVA_FILE.tmp
- cat $JAVA_FILE >> $JAVA_FILE.tmp
- mv $JAVA_FILE.tmp $JAVA_FILE
- cp $JAVA_FILE $DEST_DIR
- done
- rm -rf $JAVA_DIR
- </echo>
- <exec executable="sh" dir="${basedir}" failonerror="true">
- <arg line="target/compile-thrift.sh"/>
- </exec>
- </target>
- </configuration>
- </execution>
- </executions>
- </plugin>
- </plugins>
- </build>
- </profile>
- <profile>
- <id>nonThrift</id>
- <activation>
- <activeByDefault>true</activeByDefault>
- </activation>
- <build>
- <pluginManagement>
- <plugins>
- <plugin>
- <groupId>org.apache.maven.plugins</groupId>
- <artifactId>maven-compiler-plugin</artifactId>
- <version>${mvn-compiler-plugin.version}</version>
- <configuration>
- <source>1.6</source>
- <target>1.6</target>
- <excludes>
- <exclude>**/generated-sources/**</exclude>
- </excludes>
- </configuration>
- </plugin>
- </plugins>
- </pluginManagement>
- </build>
- </profile>
- </profiles>
-
- <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>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.apache.thrift</groupId>
- <artifactId>libthrift</artifactId>
- </dependency>
-
- <dependency>
- <groupId>org.mockito</groupId>
- <artifactId>mockito-all</artifactId>
- <scope>test</scope>
- </dependency>
-
- </dependencies>
-
-</project>
diff --git a/flume-ng-sources/flume-scribe-source/src/main/java/org/apache/flume/source/scribe/LogEntry.java b/flume-ng-sources/flume-scribe-source/src/main/java/org/apache/flume/source/scribe/LogEntry.java
deleted file mode 100644
index bd0a4c562..000000000
--- a/flume-ng-sources/flume-scribe-source/src/main/java/org/apache/flume/source/scribe/LogEntry.java
+++ /dev/null
@@ -1,521 +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.
- */
-/**
- * Autogenerated by Thrift Compiler (0.9.3)
- *
- * DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
- * @generated
- */
-package org.apache.flume.source.scribe;
-
-import org.apache.thrift.scheme.IScheme;
-import org.apache.thrift.scheme.SchemeFactory;
-import org.apache.thrift.scheme.StandardScheme;
-
-import org.apache.thrift.scheme.TupleScheme;
-import org.apache.thrift.protocol.TTupleProtocol;
-import org.apache.thrift.protocol.TProtocolException;
-import org.apache.thrift.EncodingUtils;
-import org.apache.thrift.TException;
-import org.apache.thrift.async.AsyncMethodCallback;
-import org.apache.thrift.server.AbstractNonblockingServer.*;
-import java.util.List;
-import java.util.ArrayList;
-import java.util.Map;
-import java.util.HashMap;
-import java.util.EnumMap;
-import java.util.Set;
-import java.util.HashSet;
-import java.util.EnumSet;
-import java.util.Collections;
-import java.util.BitSet;
-import java.nio.ByteBuffer;
-import java.util.Arrays;
-import javax.annotation.Generated;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
-@Generated(value = "Autogenerated by Thrift Compiler (0.9.3)", date = "2017-09-01")
-public class LogEntry implements org.apache.thrift.TBase<LogEntry, LogEntry._Fields>, java.io.Serializable, Cloneable, Comparable<LogEntry> {
- private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("LogEntry");
-
- private static final org.apache.thrift.protocol.TField CATEGORY_FIELD_DESC = new org.apache.thrift.protocol.TField("category", org.apache.thrift.protocol.TType.STRING, (short)1);
- private static final org.apache.thrift.protocol.TField MESSAGE_FIELD_DESC = new org.apache.thrift.protocol.TField("message", org.apache.thrift.protocol.TType.STRING, (short)2);
-
- private static final Map<Class<? extends IScheme>, SchemeFactory> schemes = new HashMap<Class<? extends IScheme>, SchemeFactory>();
- static {
- schemes.put(StandardScheme.class, new LogEntryStandardSchemeFactory());
- schemes.put(TupleScheme.class, new LogEntryTupleSchemeFactory());
- }
-
- public String category; // required
- public String message; // required
-
- /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
- public enum _Fields implements org.apache.thrift.TFieldIdEnum {
- CATEGORY((short)1, "category"),
- MESSAGE((short)2, "message");
-
- private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
-
- static {
- for (_Fields field : EnumSet.allOf(_Fields.class)) {
- byName.put(field.getFieldName(), field);
- }
- }
-
- /**
- * Find the _Fields constant that matches fieldId, or null if its not found.
- */
- public static _Fields findByThriftId(int fieldId) {
- switch(fieldId) {
- case 1: // CATEGORY
- return CATEGORY;
- case 2: // MESSAGE
- return MESSAGE;
- default:
- return null;
- }
- }
-
- /**
- * Find the _Fields constant that matches fieldId, throwing an exception
- * if it is not found.
- */
- public static _Fields findByThriftIdOrThrow(int fieldId) {
- _Fields fields = findByThriftId(fieldId);
- if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!");
- return fields;
- }
-
- /**
- * Find the _Fields constant that matches name, or null if its not found.
- */
- public static _Fields findByName(String name) {
- return byName.get(name);
- }
-
- private final short _thriftId;
- private final String _fieldName;
-
- _Fields(short thriftId, String fieldName) {
- _thriftId = thriftId;
- _fieldName = fieldName;
- }
-
- public short getThriftFieldId() {
- return _thriftId;
- }
-
- public String getFieldName() {
- return _fieldName;
- }
- }
-
- // isset id assignments
- public static final Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
- static {
- Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
- tmpMap.put(_Fields.CATEGORY, new org.apache.thrift.meta_data.FieldMetaData("category", org.apache.thrift.TFieldRequirementType.DEFAULT,
- new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING)));
- tmpMap.put(_Fields.MESSAGE, new org.apache.thrift.meta_data.FieldMetaData("message", org.apache.thrift.TFieldRequirementType.DEFAULT,
- new org.apache.thrift.meta_data.FieldValueMetaData(org.apache.thrift.protocol.TType.STRING)));
- metaDataMap = Collections.unmodifiableMap(tmpMap);
- org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(LogEntry.class, metaDataMap);
- }
-
- public LogEntry() {
- }
-
- public LogEntry(
- String category,
- String message)
- {
- this();
- this.category = category;
- this.message = message;
- }
-
- /**
- * Performs a deep copy on <i>other</i>.
- */
- public LogEntry(LogEntry other) {
- if (other.isSetCategory()) {
- this.category = other.category;
- }
- if (other.isSetMessage()) {
- this.message = other.message;
- }
- }
-
- public LogEntry deepCopy() {
- return new LogEntry(this);
- }
-
- @Override
- public void clear() {
- this.category = null;
- this.message = null;
- }
-
- public String getCategory() {
- return this.category;
- }
-
- public LogEntry setCategory(String category) {
- this.category = category;
- return this;
- }
-
- public void unsetCategory() {
- this.category = null;
- }
-
- /** Returns true if field category is set (has been assigned a value) and false otherwise */
- public boolean isSetCategory() {
- return this.category != null;
- }
-
- public void setCategoryIsSet(boolean value) {
- if (!value) {
- this.category = null;
- }
- }
-
- public String getMessage() {
- return this.message;
- }
-
- public LogEntry setMessage(String message) {
- this.message = message;
- return this;
- }
-
- public void unsetMessage() {
- this.message = null;
- }
-
- /** Returns true if field message is set (has been assigned a value) and false otherwise */
- public boolean isSetMessage() {
- return this.message != null;
- }
-
- public void setMessageIsSet(boolean value) {
- if (!value) {
- this.message = null;
- }
- }
-
- public void setFieldValue(_Fields field, Object value) {
- switch (field) {
- case CATEGORY:
- if (value == null) {
- unsetCategory();
- } else {
- setCategory((String)value);
- }
- break;
-
- case MESSAGE:
- if (value == null) {
- unsetMessage();
- } else {
- setMessage((String)value);
- }
- break;
-
- }
- }
-
- public Object getFieldValue(_Fields field) {
- switch (field) {
- case CATEGORY:
- return getCategory();
-
- case MESSAGE:
- return getMessage();
-
- }
- throw new IllegalStateException();
- }
-
- /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
- public boolean isSet(_Fields field) {
- if (field == null) {
- throw new IllegalArgumentException();
- }
-
- switch (field) {
- case CATEGORY:
- return isSetCategory();
- case MESSAGE:
- return isSetMessage();
- }
- throw new IllegalStateException();
- }
-
- @Override
- public boolean equals(Object that) {
- if (that == null)
- return false;
- if (that instanceof LogEntry)
- return this.equals((LogEntry)that);
- return false;
- }
-
- public boolean equals(LogEntry that) {
- if (that == null)
- return false;
-
- boolean this_present_category = true && this.isSetCategory();
- boolean that_present_category = true && that.isSetCategory();
- if (this_present_category || that_present_category) {
- if (!(this_present_category && that_present_category))
- return false;
- if (!this.category.equals(that.category))
- return false;
- }
-
- boolean this_present_message = true && this.isSetMessage();
- boolean that_present_message = true && that.isSetMessage();
- if (this_present_message || that_present_message) {
- if (!(this_present_message && that_present_message))
- return false;
- if (!this.message.equals(that.message))
- return false;
- }
-
- return true;
- }
-
- @Override
- public int hashCode() {
- List<Object> list = new ArrayList<Object>();
-
- boolean present_category = true && (isSetCategory());
- list.add(present_category);
- if (present_category)
- list.add(category);
-
- boolean present_message = true && (isSetMessage());
- list.add(present_message);
- if (present_message)
- list.add(message);
-
- return list.hashCode();
- }
-
- @Override
- public int compareTo(LogEntry other) {
- if (!getClass().equals(other.getClass())) {
- return getClass().getName().compareTo(other.getClass().getName());
- }
-
- int lastComparison = 0;
-
- lastComparison = Boolean.valueOf(isSetCategory()).compareTo(other.isSetCategory());
- if (lastComparison != 0) {
- return lastComparison;
- }
- if (isSetCategory()) {
- lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.category, other.category);
- if (lastComparison != 0) {
- return lastComparison;
- }
- }
- lastComparison = Boolean.valueOf(isSetMessage()).compareTo(other.isSetMessage());
- if (lastComparison != 0) {
- return lastComparison;
- }
- if (isSetMessage()) {
- lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.message, other.message);
- if (lastComparison != 0) {
- return lastComparison;
- }
- }
- return 0;
- }
-
- public _Fields fieldForId(int fieldId) {
- return _Fields.findByThriftId(fieldId);
- }
-
- public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
- schemes.get(iprot.getScheme()).getScheme().read(iprot, this);
- }
-
- public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
- schemes.get(oprot.getScheme()).getScheme().write(oprot, this);
- }
-
- @Override
- public String toString() {
- StringBuilder sb = new StringBuilder("LogEntry(");
- boolean first = true;
-
- sb.append("category:");
- if (this.category == null) {
- sb.append("null");
- } else {
- sb.append(this.category);
- }
- first = false;
- if (!first) sb.append(", ");
- sb.append("message:");
- if (this.message == null) {
- sb.append("null");
- } else {
- sb.append(this.message);
- }
- first = false;
- sb.append(")");
- return sb.toString();
- }
-
- public void validate() throws org.apache.thrift.TException {
- // check for required fields
- // check for sub-struct validity
- }
-
- private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
- try {
- write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
- } catch (org.apache.thrift.TException te) {
- throw new java.io.IOException(te);
- }
- }
-
- private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
- try {
- read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
- } catch (org.apache.thrift.TException te) {
- throw new java.io.IOException(te);
- }
- }
-
- private static class LogEntryStandardSchemeFactory implements SchemeFactory {
- public LogEntryStandardScheme getScheme() {
- return new LogEntryStandardScheme();
- }
- }
-
- private static class LogEntryStandardScheme extends StandardScheme<LogEntry> {
-
- public void read(org.apache.thrift.protocol.TProtocol iprot, LogEntry struct) throws org.apache.thrift.TException {
- org.apache.thrift.protocol.TField schemeField;
- iprot.readStructBegin();
- while (true)
- {
- schemeField = iprot.readFieldBegin();
- if (schemeField.type == org.apache.thrift.protocol.TType.STOP) {
- break;
- }
- switch (schemeField.id) {
- case 1: // CATEGORY
- if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
- struct.category = iprot.readString();
- struct.setCategoryIsSet(true);
- } else {
- org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
- }
- break;
- case 2: // MESSAGE
- if (schemeField.type == org.apache.thrift.protocol.TType.STRING) {
- struct.message = iprot.readString();
- struct.setMessageIsSet(true);
- } else {
- org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
- }
- break;
- default:
- org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
- }
- iprot.readFieldEnd();
- }
- iprot.readStructEnd();
-
- // check for required fields of primitive type, which can't be checked in the validate method
- struct.validate();
- }
-
- public void write(org.apache.thrift.protocol.TProtocol oprot, LogEntry struct) throws org.apache.thrift.TException {
- struct.validate();
-
- oprot.writeStructBegin(STRUCT_DESC);
- if (struct.category != null) {
- oprot.writeFieldBegin(CATEGORY_FIELD_DESC);
- oprot.writeString(struct.category);
- oprot.writeFieldEnd();
- }
- if (struct.message != null) {
- oprot.writeFieldBegin(MESSAGE_FIELD_DESC);
- oprot.writeString(struct.message);
- oprot.writeFieldEnd();
- }
- oprot.writeFieldStop();
- oprot.writeStructEnd();
- }
-
- }
-
- private static class LogEntryTupleSchemeFactory implements SchemeFactory {
- public LogEntryTupleScheme getScheme() {
- return new LogEntryTupleScheme();
- }
- }
-
- private static class LogEntryTupleScheme extends TupleScheme<LogEntry> {
-
- @Override
- public void write(org.apache.thrift.protocol.TProtocol prot, LogEntry struct) throws org.apache.thrift.TException {
- TTupleProtocol oprot = (TTupleProtocol) prot;
- BitSet optionals = new BitSet();
- if (struct.isSetCategory()) {
- optionals.set(0);
- }
- if (struct.isSetMessage()) {
- optionals.set(1);
- }
- oprot.writeBitSet(optionals, 2);
- if (struct.isSetCategory()) {
- oprot.writeString(struct.category);
- }
- if (struct.isSetMessage()) {
- oprot.writeString(struct.message);
- }
- }
-
- @Override
- public void read(org.apache.thrift.protocol.TProtocol prot, LogEntry struct) throws org.apache.thrift.TException {
- TTupleProtocol iprot = (TTupleProtocol) prot;
- BitSet incoming = iprot.readBitSet(2);
- if (incoming.get(0)) {
- struct.category = iprot.readString();
- struct.setCategoryIsSet(true);
- }
- if (incoming.get(1)) {
- struct.message = iprot.readString();
- struct.setMessageIsSet(true);
- }
- }
- }
-
-}
-
diff --git a/flume-ng-sources/flume-scribe-source/src/main/java/org/apache/flume/source/scribe/ResultCode.java b/flume-ng-sources/flume-scribe-source/src/main/java/org/apache/flume/source/scribe/ResultCode.java
deleted file mode 100644
index fc7d773e2..000000000
--- a/flume-ng-sources/flume-scribe-source/src/main/java/org/apache/flume/source/scribe/ResultCode.java
+++ /dev/null
@@ -1,63 +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.
- */
-/**
- * Autogenerated by Thrift Compiler (0.9.3)
- *
- * DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
- * @generated
- */
-package org.apache.flume.source.scribe;
-
-
-import java.util.Map;
-import java.util.HashMap;
-import org.apache.thrift.TEnum;
-
-public enum ResultCode implements org.apache.thrift.TEnum {
- OK(0),
- TRY_LATER(1);
-
- private final int value;
-
- private ResultCode(int value) {
- this.value = value;
- }
-
- /**
- * Get the integer value of this enum value, as defined in the Thrift IDL.
- */
- public int getValue() {
- return value;
- }
-
- /**
- * Find a the enum type by its integer value, as defined in the Thrift IDL.
- * @return null if the value is not found.
- */
- public static ResultCode findByValue(int value) {
- switch (value) {
- case 0:
- return OK;
- case 1:
- return TRY_LATER;
- default:
- return null;
- }
- }
-}
diff --git a/flume-ng-sources/flume-scribe-source/src/main/java/org/apache/flume/source/scribe/Scribe.java b/flume-ng-sources/flume-scribe-source/src/main/java/org/apache/flume/source/scribe/Scribe.java
deleted file mode 100644
index d7723db77..000000000
--- a/flume-ng-sources/flume-scribe-source/src/main/java/org/apache/flume/source/scribe/Scribe.java
+++ /dev/null
@@ -1,1056 +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.
- */
-/**
- * Autogenerated by Thrift Compiler (0.9.3)
- *
- * DO NOT EDIT UNLESS YOU ARE SURE THAT YOU KNOW WHAT YOU ARE DOING
- * @generated
- */
-package org.apache.flume.source.scribe;
-
-import org.apache.thrift.scheme.IScheme;
-import org.apache.thrift.scheme.SchemeFactory;
-import org.apache.thrift.scheme.StandardScheme;
-
-import org.apache.thrift.scheme.TupleScheme;
-import org.apache.thrift.protocol.TTupleProtocol;
-import org.apache.thrift.TException;
-import org.apache.thrift.async.AsyncMethodCallback;
-import org.apache.thrift.server.AbstractNonblockingServer;
-import java.util.List;
-import java.util.ArrayList;
-import java.util.Map;
-import java.util.HashMap;
-import java.util.EnumMap;
-import java.util.EnumSet;
-import java.util.Collections;
-import java.util.BitSet;
-import javax.annotation.Generated;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-@SuppressWarnings({"cast", "rawtypes", "serial", "unchecked"})
-@Generated(value = "Autogenerated by Thrift Compiler (0.9.3)", date = "2017-09-01")
-public class Scribe {
-
- public interface Iface {
-
- public ResultCode Log(List<LogEntry> messages) throws org.apache.thrift.TException;
-
- }
-
- public interface AsyncIface {
-
- public void Log(List<LogEntry> messages, org.apache.thrift.async.AsyncMethodCallback resultHandler) throws org.apache.thrift.TException;
-
- }
-
- public static class Client extends org.apache.thrift.TServiceClient implements Iface {
- public static class Factory implements org.apache.thrift.TServiceClientFactory<Client> {
- public Factory() {}
- public Client getClient(org.apache.thrift.protocol.TProtocol prot) {
- return new Client(prot);
- }
- public Client getClient(org.apache.thrift.protocol.TProtocol iprot, org.apache.thrift.protocol.TProtocol oprot) {
- return new Client(iprot, oprot);
- }
- }
-
- public Client(org.apache.thrift.protocol.TProtocol prot)
- {
- super(prot, prot);
- }
-
- public Client(org.apache.thrift.protocol.TProtocol iprot, org.apache.thrift.protocol.TProtocol oprot) {
- super(iprot, oprot);
- }
-
- public ResultCode Log(List<LogEntry> messages) throws org.apache.thrift.TException
- {
- send_Log(messages);
- return recv_Log();
- }
-
- public void send_Log(List<LogEntry> messages) throws org.apache.thrift.TException
- {
- Log_args args = new Log_args();
- args.setMessages(messages);
- sendBase("Log", args);
- }
-
- public ResultCode recv_Log() throws org.apache.thrift.TException
- {
- Log_result result = new Log_result();
- receiveBase(result, "Log");
- if (result.isSetSuccess()) {
- return result.success;
- }
- throw new org.apache.thrift.TApplicationException(org.apache.thrift.TApplicationException.MISSING_RESULT, "Log failed: unknown result");
- }
-
- }
- public static class AsyncClient extends org.apache.thrift.async.TAsyncClient implements AsyncIface {
- public static class Factory implements org.apache.thrift.async.TAsyncClientFactory<AsyncClient> {
- private org.apache.thrift.async.TAsyncClientManager clientManager;
- private org.apache.thrift.protocol.TProtocolFactory protocolFactory;
- public Factory(org.apache.thrift.async.TAsyncClientManager clientManager, org.apache.thrift.protocol.TProtocolFactory protocolFactory) {
- this.clientManager = clientManager;
- this.protocolFactory = protocolFactory;
- }
- public AsyncClient getAsyncClient(org.apache.thrift.transport.TNonblockingTransport transport) {
- return new AsyncClient(protocolFactory, clientManager, transport);
- }
- }
-
- public AsyncClient(org.apache.thrift.protocol.TProtocolFactory protocolFactory, org.apache.thrift.async.TAsyncClientManager clientManager, org.apache.thrift.transport.TNonblockingTransport transport) {
- super(protocolFactory, clientManager, transport);
- }
-
- public void Log(List<LogEntry> messages, org.apache.thrift.async.AsyncMethodCallback resultHandler) throws org.apache.thrift.TException {
- checkReady();
- Log_call method_call = new Log_call(messages, resultHandler, this, ___protocolFactory, ___transport);
- this.___currentMethod = method_call;
- ___manager.call(method_call);
- }
-
- public static class Log_call extends org.apache.thrift.async.TAsyncMethodCall {
- private List<LogEntry> messages;
- public Log_call(List<LogEntry> messages, org.apache.thrift.async.AsyncMethodCallback resultHandler, org.apache.thrift.async.TAsyncClient client, org.apache.thrift.protocol.TProtocolFactory protocolFactory, org.apache.thrift.transport.TNonblockingTransport transport) throws org.apache.thrift.TException {
- super(client, protocolFactory, transport, resultHandler, false);
- this.messages = messages;
- }
-
- public void write_args(org.apache.thrift.protocol.TProtocol prot) throws org.apache.thrift.TException {
- prot.writeMessageBegin(new org.apache.thrift.protocol.TMessage("Log", org.apache.thrift.protocol.TMessageType.CALL, 0));
- Log_args args = new Log_args();
- args.setMessages(messages);
- args.write(prot);
- prot.writeMessageEnd();
- }
-
- public ResultCode getResult() throws org.apache.thrift.TException {
- if (getState() != org.apache.thrift.async.TAsyncMethodCall.State.RESPONSE_READ) {
- throw new IllegalStateException("Method call not finished!");
- }
- org.apache.thrift.transport.TMemoryInputTransport memoryTransport = new org.apache.thrift.transport.TMemoryInputTransport(getFrameBuffer().array());
- org.apache.thrift.protocol.TProtocol prot = client.getProtocolFactory().getProtocol(memoryTransport);
- return (new Client(prot)).recv_Log();
- }
- }
-
- }
-
- public static class Processor<I extends Iface> extends org.apache.thrift.TBaseProcessor<I> implements org.apache.thrift.TProcessor {
- private static final Logger LOGGER = LoggerFactory.getLogger(Processor.class.getName());
- public Processor(I iface) {
- super(iface, getProcessMap(new HashMap<String, org.apache.thrift.ProcessFunction<I, ? extends org.apache.thrift.TBase>>()));
- }
-
- protected Processor(I iface, Map<String, org.apache.thrift.ProcessFunction<I, ? extends org.apache.thrift.TBase>> processMap) {
- super(iface, getProcessMap(processMap));
- }
-
- private static <I extends Iface> Map<String, org.apache.thrift.ProcessFunction<I, ? extends org.apache.thrift.TBase>> getProcessMap(Map<String, org.apache.thrift.ProcessFunction<I, ? extends org.apache.thrift.TBase>> processMap) {
- processMap.put("Log", new Log());
- return processMap;
- }
-
- public static class Log<I extends Iface> extends org.apache.thrift.ProcessFunction<I, Log_args> {
- public Log() {
- super("Log");
- }
-
- public Log_args getEmptyArgsInstance() {
- return new Log_args();
- }
-
- protected boolean isOneway() {
- return false;
- }
-
- public Log_result getResult(I iface, Log_args args) throws org.apache.thrift.TException {
- Log_result result = new Log_result();
- result.success = iface.Log(args.messages);
- return result;
- }
- }
-
- }
-
- public static class AsyncProcessor<I extends AsyncIface> extends org.apache.thrift.TBaseAsyncProcessor<I> {
- private static final Logger LOGGER = LoggerFactory.getLogger(AsyncProcessor.class.getName());
- public AsyncProcessor(I iface) {
- super(iface, getProcessMap(new HashMap<String, org.apache.thrift.AsyncProcessFunction<I, ? extends org.apache.thrift.TBase, ?>>()));
- }
-
- protected AsyncProcessor(I iface, Map<String, org.apache.thrift.AsyncProcessFunction<I, ? extends org.apache.thrift.TBase, ?>> processMap) {
- super(iface, getProcessMap(processMap));
- }
-
- private static <I extends AsyncIface> Map<String, org.apache.thrift.AsyncProcessFunction<I, ? extends org.apache.thrift.TBase,?>> getProcessMap(Map<String, org.apache.thrift.AsyncProcessFunction<I, ? extends org.apache.thrift.TBase, ?>> processMap) {
- processMap.put("Log", new Log());
- return processMap;
- }
-
- public static class Log<I extends AsyncIface> extends org.apache.thrift.AsyncProcessFunction<I, Log_args, ResultCode> {
- public Log() {
- super("Log");
- }
-
- public Log_args getEmptyArgsInstance() {
- return new Log_args();
- }
-
- public AsyncMethodCallback<ResultCode> getResultHandler(final AbstractNonblockingServer.AsyncFrameBuffer fb, final int seqid) {
- final org.apache.thrift.AsyncProcessFunction fcall = this;
- return new AsyncMethodCallback<ResultCode>() {
- public void onComplete(ResultCode o) {
- Log_result result = new Log_result();
- result.success = o;
- try {
- fcall.sendResponse(fb,result, org.apache.thrift.protocol.TMessageType.REPLY,seqid);
- return;
- } catch (Exception e) {
- LOGGER.error("Exception writing to internal frame buffer", e);
- }
- fb.close();
- }
- public void onError(Exception e) {
- byte msgType = org.apache.thrift.protocol.TMessageType.REPLY;
- org.apache.thrift.TBase msg;
- Log_result result = new Log_result();
- {
- msgType = org.apache.thrift.protocol.TMessageType.EXCEPTION;
- msg = (org.apache.thrift.TBase)new org.apache.thrift.TApplicationException(org.apache.thrift.TApplicationException.INTERNAL_ERROR, e.getMessage());
- }
- try {
- fcall.sendResponse(fb,msg,msgType,seqid);
- return;
- } catch (Exception ex) {
- LOGGER.error("Exception writing to internal frame buffer", ex);
- }
- fb.close();
- }
- };
- }
-
- protected boolean isOneway() {
- return false;
- }
-
- public void start(I iface, Log_args args, org.apache.thrift.async.AsyncMethodCallback<ResultCode> resultHandler) throws TException {
- iface.Log(args.messages,resultHandler);
- }
- }
-
- }
-
- public static class Log_args implements org.apache.thrift.TBase<Log_args, Log_args._Fields>, java.io.Serializable, Cloneable, Comparable<Log_args> {
- private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("Log_args");
-
- private static final org.apache.thrift.protocol.TField MESSAGES_FIELD_DESC = new org.apache.thrift.protocol.TField("messages", org.apache.thrift.protocol.TType.LIST, (short)1);
-
- private static final Map<Class<? extends IScheme>, SchemeFactory> schemes = new HashMap<Class<? extends IScheme>, SchemeFactory>();
- static {
- schemes.put(StandardScheme.class, new Log_argsStandardSchemeFactory());
- schemes.put(TupleScheme.class, new Log_argsTupleSchemeFactory());
- }
-
- public List<LogEntry> messages; // required
-
- /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
- public enum _Fields implements org.apache.thrift.TFieldIdEnum {
- MESSAGES((short)1, "messages");
-
- private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
-
- static {
- for (_Fields field : EnumSet.allOf(_Fields.class)) {
- byName.put(field.getFieldName(), field);
- }
- }
-
- /**
- * Find the _Fields constant that matches fieldId, or null if its not found.
- */
- public static _Fields findByThriftId(int fieldId) {
- switch(fieldId) {
- case 1: // MESSAGES
- return MESSAGES;
- default:
- return null;
- }
- }
-
- /**
- * Find the _Fields constant that matches fieldId, throwing an exception
- * if it is not found.
- */
- public static _Fields findByThriftIdOrThrow(int fieldId) {
- _Fields fields = findByThriftId(fieldId);
- if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!");
- return fields;
- }
-
- /**
- * Find the _Fields constant that matches name, or null if its not found.
- */
- public static _Fields findByName(String name) {
- return byName.get(name);
- }
-
- private final short _thriftId;
- private final String _fieldName;
-
- _Fields(short thriftId, String fieldName) {
- _thriftId = thriftId;
- _fieldName = fieldName;
- }
-
- public short getThriftFieldId() {
- return _thriftId;
- }
-
- public String getFieldName() {
- return _fieldName;
- }
- }
-
- // isset id assignments
- public static final Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
- static {
- Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
- tmpMap.put(_Fields.MESSAGES, new org.apache.thrift.meta_data.FieldMetaData("messages", org.apache.thrift.TFieldRequirementType.DEFAULT,
- new org.apache.thrift.meta_data.ListMetaData(org.apache.thrift.protocol.TType.LIST,
- new org.apache.thrift.meta_data.StructMetaData(org.apache.thrift.protocol.TType.STRUCT, LogEntry.class))));
- metaDataMap = Collections.unmodifiableMap(tmpMap);
- org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(Log_args.class, metaDataMap);
- }
-
- public Log_args() {
- }
-
- public Log_args(
- List<LogEntry> messages)
- {
- this();
- this.messages = messages;
- }
-
- /**
- * Performs a deep copy on <i>other</i>.
- */
- public Log_args(Log_args other) {
- if (other.isSetMessages()) {
- List<LogEntry> __this__messages = new ArrayList<LogEntry>(other.messages.size());
- for (LogEntry other_element : other.messages) {
- __this__messages.add(new LogEntry(other_element));
- }
- this.messages = __this__messages;
- }
- }
-
- public Log_args deepCopy() {
- return new Log_args(this);
- }
-
- @Override
- public void clear() {
- this.messages = null;
- }
-
- public int getMessagesSize() {
- return (this.messages == null) ? 0 : this.messages.size();
- }
-
- public java.util.Iterator<LogEntry> getMessagesIterator() {
- return (this.messages == null) ? null : this.messages.iterator();
- }
-
- public void addToMessages(LogEntry elem) {
- if (this.messages == null) {
- this.messages = new ArrayList<LogEntry>();
- }
- this.messages.add(elem);
- }
-
- public List<LogEntry> getMessages() {
- return this.messages;
- }
-
- public Log_args setMessages(List<LogEntry> messages) {
- this.messages = messages;
- return this;
- }
-
- public void unsetMessages() {
- this.messages = null;
- }
-
- /** Returns true if field messages is set (has been assigned a value) and false otherwise */
- public boolean isSetMessages() {
- return this.messages != null;
- }
-
- public void setMessagesIsSet(boolean value) {
- if (!value) {
- this.messages = null;
- }
- }
-
- public void setFieldValue(_Fields field, Object value) {
- switch (field) {
- case MESSAGES:
- if (value == null) {
- unsetMessages();
- } else {
- setMessages((List<LogEntry>)value);
- }
- break;
-
- }
- }
-
- public Object getFieldValue(_Fields field) {
- switch (field) {
- case MESSAGES:
- return getMessages();
-
- }
- throw new IllegalStateException();
- }
-
- /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
- public boolean isSet(_Fields field) {
- if (field == null) {
- throw new IllegalArgumentException();
- }
-
- switch (field) {
- case MESSAGES:
- return isSetMessages();
- }
- throw new IllegalStateException();
- }
-
- @Override
- public boolean equals(Object that) {
- if (that == null)
- return false;
- if (that instanceof Log_args)
- return this.equals((Log_args)that);
- return false;
- }
-
- public boolean equals(Log_args that) {
- if (that == null)
- return false;
-
- boolean this_present_messages = true && this.isSetMessages();
- boolean that_present_messages = true && that.isSetMessages();
- if (this_present_messages || that_present_messages) {
- if (!(this_present_messages && that_present_messages))
- return false;
- if (!this.messages.equals(that.messages))
- return false;
- }
-
- return true;
- }
-
- @Override
- public int hashCode() {
- List<Object> list = new ArrayList<Object>();
-
- boolean present_messages = true && (isSetMessages());
- list.add(present_messages);
- if (present_messages)
- list.add(messages);
-
- return list.hashCode();
- }
-
- @Override
- public int compareTo(Log_args other) {
- if (!getClass().equals(other.getClass())) {
- return getClass().getName().compareTo(other.getClass().getName());
- }
-
- int lastComparison = 0;
-
- lastComparison = Boolean.valueOf(isSetMessages()).compareTo(other.isSetMessages());
- if (lastComparison != 0) {
- return lastComparison;
- }
- if (isSetMessages()) {
- lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.messages, other.messages);
- if (lastComparison != 0) {
- return lastComparison;
- }
- }
- return 0;
- }
-
- public _Fields fieldForId(int fieldId) {
- return _Fields.findByThriftId(fieldId);
- }
-
- public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
- schemes.get(iprot.getScheme()).getScheme().read(iprot, this);
- }
-
- public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
- schemes.get(oprot.getScheme()).getScheme().write(oprot, this);
- }
-
- @Override
- public String toString() {
- StringBuilder sb = new StringBuilder("Log_args(");
- boolean first = true;
-
- sb.append("messages:");
- if (this.messages == null) {
- sb.append("null");
- } else {
- sb.append(this.messages);
- }
- first = false;
- sb.append(")");
- return sb.toString();
- }
-
- public void validate() throws org.apache.thrift.TException {
- // check for required fields
- // check for sub-struct validity
- }
-
- private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
- try {
- write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
- } catch (org.apache.thrift.TException te) {
- throw new java.io.IOException(te);
- }
- }
-
- private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
- try {
- read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
- } catch (org.apache.thrift.TException te) {
- throw new java.io.IOException(te);
- }
- }
-
- private static class Log_argsStandardSchemeFactory implements SchemeFactory {
- public Log_argsStandardScheme getScheme() {
- return new Log_argsStandardScheme();
- }
- }
-
- private static class Log_argsStandardScheme extends StandardScheme<Log_args> {
-
- public void read(org.apache.thrift.protocol.TProtocol iprot, Log_args struct) throws org.apache.thrift.TException {
- org.apache.thrift.protocol.TField schemeField;
- iprot.readStructBegin();
- while (true)
- {
- schemeField = iprot.readFieldBegin();
- if (schemeField.type == org.apache.thrift.protocol.TType.STOP) {
- break;
- }
- switch (schemeField.id) {
- case 1: // MESSAGES
- if (schemeField.type == org.apache.thrift.protocol.TType.LIST) {
- {
- org.apache.thrift.protocol.TList _list0 = iprot.readListBegin();
- struct.messages = new ArrayList<LogEntry>(_list0.size);
- LogEntry _elem1;
- for (int _i2 = 0; _i2 < _list0.size; ++_i2)
- {
- _elem1 = new LogEntry();
- _elem1.read(iprot);
- struct.messages.add(_elem1);
- }
- iprot.readListEnd();
- }
- struct.setMessagesIsSet(true);
- } else {
- org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
- }
- break;
- default:
- org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
- }
- iprot.readFieldEnd();
- }
- iprot.readStructEnd();
-
- // check for required fields of primitive type, which can't be checked in the validate method
- struct.validate();
- }
-
- public void write(org.apache.thrift.protocol.TProtocol oprot, Log_args struct) throws org.apache.thrift.TException {
- struct.validate();
-
- oprot.writeStructBegin(STRUCT_DESC);
- if (struct.messages != null) {
- oprot.writeFieldBegin(MESSAGES_FIELD_DESC);
- {
- oprot.writeListBegin(new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, struct.messages.size()));
- for (LogEntry _iter3 : struct.messages)
- {
- _iter3.write(oprot);
- }
- oprot.writeListEnd();
- }
- oprot.writeFieldEnd();
- }
- oprot.writeFieldStop();
- oprot.writeStructEnd();
- }
-
- }
-
- private static class Log_argsTupleSchemeFactory implements SchemeFactory {
- public Log_argsTupleScheme getScheme() {
- return new Log_argsTupleScheme();
- }
- }
-
- private static class Log_argsTupleScheme extends TupleScheme<Log_args> {
-
- @Override
- public void write(org.apache.thrift.protocol.TProtocol prot, Log_args struct) throws org.apache.thrift.TException {
- TTupleProtocol oprot = (TTupleProtocol) prot;
- BitSet optionals = new BitSet();
- if (struct.isSetMessages()) {
- optionals.set(0);
- }
- oprot.writeBitSet(optionals, 1);
- if (struct.isSetMessages()) {
- {
- oprot.writeI32(struct.messages.size());
- for (LogEntry _iter4 : struct.messages)
- {
- _iter4.write(oprot);
- }
- }
- }
- }
-
- @Override
- public void read(org.apache.thrift.protocol.TProtocol prot, Log_args struct) throws org.apache.thrift.TException {
- TTupleProtocol iprot = (TTupleProtocol) prot;
- BitSet incoming = iprot.readBitSet(1);
- if (incoming.get(0)) {
- {
- org.apache.thrift.protocol.TList _list5 = new org.apache.thrift.protocol.TList(org.apache.thrift.protocol.TType.STRUCT, iprot.readI32());
- struct.messages = new ArrayList<LogEntry>(_list5.size);
- LogEntry _elem6;
- for (int _i7 = 0; _i7 < _list5.size; ++_i7)
- {
- _elem6 = new LogEntry();
- _elem6.read(iprot);
- struct.messages.add(_elem6);
- }
- }
- struct.setMessagesIsSet(true);
- }
- }
- }
-
- }
-
- public static class Log_result implements org.apache.thrift.TBase<Log_result, Log_result._Fields>, java.io.Serializable, Cloneable, Comparable<Log_result> {
- private static final org.apache.thrift.protocol.TStruct STRUCT_DESC = new org.apache.thrift.protocol.TStruct("Log_result");
-
- private static final org.apache.thrift.protocol.TField SUCCESS_FIELD_DESC = new org.apache.thrift.protocol.TField("success", org.apache.thrift.protocol.TType.I32, (short)0);
-
- private static final Map<Class<? extends IScheme>, SchemeFactory> schemes = new HashMap<Class<? extends IScheme>, SchemeFactory>();
- static {
- schemes.put(StandardScheme.class, new Log_resultStandardSchemeFactory());
- schemes.put(TupleScheme.class, new Log_resultTupleSchemeFactory());
- }
-
- /**
- *
- * @see ResultCode
- */
- public ResultCode success; // required
-
- /** The set of fields this struct contains, along with convenience methods for finding and manipulating them. */
- public enum _Fields implements org.apache.thrift.TFieldIdEnum {
- /**
- *
- * @see ResultCode
- */
- SUCCESS((short)0, "success");
-
- private static final Map<String, _Fields> byName = new HashMap<String, _Fields>();
-
- static {
- for (_Fields field : EnumSet.allOf(_Fields.class)) {
- byName.put(field.getFieldName(), field);
- }
- }
-
- /**
- * Find the _Fields constant that matches fieldId, or null if its not found.
- */
- public static _Fields findByThriftId(int fieldId) {
- switch(fieldId) {
- case 0: // SUCCESS
- return SUCCESS;
- default:
- return null;
- }
- }
-
- /**
- * Find the _Fields constant that matches fieldId, throwing an exception
- * if it is not found.
- */
- public static _Fields findByThriftIdOrThrow(int fieldId) {
- _Fields fields = findByThriftId(fieldId);
- if (fields == null) throw new IllegalArgumentException("Field " + fieldId + " doesn't exist!");
- return fields;
- }
-
- /**
- * Find the _Fields constant that matches name, or null if its not found.
- */
- public static _Fields findByName(String name) {
- return byName.get(name);
- }
-
- private final short _thriftId;
- private final String _fieldName;
-
- _Fields(short thriftId, String fieldName) {
- _thriftId = thriftId;
- _fieldName = fieldName;
- }
-
- public short getThriftFieldId() {
- return _thriftId;
- }
-
- public String getFieldName() {
- return _fieldName;
- }
- }
-
- // isset id assignments
- public static final Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> metaDataMap;
- static {
- Map<_Fields, org.apache.thrift.meta_data.FieldMetaData> tmpMap = new EnumMap<_Fields, org.apache.thrift.meta_data.FieldMetaData>(_Fields.class);
- tmpMap.put(_Fields.SUCCESS, new org.apache.thrift.meta_data.FieldMetaData("success", org.apache.thrift.TFieldRequirementType.DEFAULT,
- new org.apache.thrift.meta_data.EnumMetaData(org.apache.thrift.protocol.TType.ENUM, ResultCode.class)));
- metaDataMap = Collections.unmodifiableMap(tmpMap);
- org.apache.thrift.meta_data.FieldMetaData.addStructMetaDataMap(Log_result.class, metaDataMap);
- }
-
- public Log_result() {
- }
-
- public Log_result(
- ResultCode success)
- {
- this();
- this.success = success;
- }
-
- /**
- * Performs a deep copy on <i>other</i>.
- */
- public Log_result(Log_result other) {
- if (other.isSetSuccess()) {
- this.success = other.success;
- }
- }
-
- public Log_result deepCopy() {
- return new Log_result(this);
- }
-
- @Override
- public void clear() {
- this.success = null;
- }
-
- /**
- *
- * @see ResultCode
- */
- public ResultCode getSuccess() {
- return this.success;
- }
-
- /**
- *
- * @see ResultCode
- */
- public Log_result setSuccess(ResultCode success) {
- this.success = success;
- return this;
- }
-
- public void unsetSuccess() {
- this.success = null;
- }
-
- /** Returns true if field success is set (has been assigned a value) and false otherwise */
- public boolean isSetSuccess() {
- return this.success != null;
- }
-
- public void setSuccessIsSet(boolean value) {
- if (!value) {
- this.success = null;
- }
- }
-
- public void setFieldValue(_Fields field, Object value) {
- switch (field) {
- case SUCCESS:
- if (value == null) {
- unsetSuccess();
- } else {
- setSuccess((ResultCode)value);
- }
- break;
-
- }
- }
-
- public Object getFieldValue(_Fields field) {
- switch (field) {
- case SUCCESS:
- return getSuccess();
-
- }
- throw new IllegalStateException();
- }
-
- /** Returns true if field corresponding to fieldID is set (has been assigned a value) and false otherwise */
- public boolean isSet(_Fields field) {
- if (field == null) {
- throw new IllegalArgumentException();
- }
-
- switch (field) {
- case SUCCESS:
- return isSetSuccess();
- }
- throw new IllegalStateException();
- }
-
- @Override
- public boolean equals(Object that) {
- if (that == null)
- return false;
- if (that instanceof Log_result)
- return this.equals((Log_result)that);
- return false;
- }
-
- public boolean equals(Log_result that) {
- if (that == null)
- return false;
-
- boolean this_present_success = true && this.isSetSuccess();
- boolean that_present_success = true && that.isSetSuccess();
- if (this_present_success || that_present_success) {
- if (!(this_present_success && that_present_success))
- return false;
- if (!this.success.equals(that.success))
- return false;
- }
-
- return true;
- }
-
- @Override
- public int hashCode() {
- List<Object> list = new ArrayList<Object>();
-
- boolean present_success = true && (isSetSuccess());
- list.add(present_success);
- if (present_success)
- list.add(success.getValue());
-
- return list.hashCode();
- }
-
- @Override
- public int compareTo(Log_result other) {
- if (!getClass().equals(other.getClass())) {
- return getClass().getName().compareTo(other.getClass().getName());
- }
-
- int lastComparison = 0;
-
- lastComparison = Boolean.valueOf(isSetSuccess()).compareTo(other.isSetSuccess());
- if (lastComparison != 0) {
- return lastComparison;
- }
- if (isSetSuccess()) {
- lastComparison = org.apache.thrift.TBaseHelper.compareTo(this.success, other.success);
- if (lastComparison != 0) {
- return lastComparison;
- }
- }
- return 0;
- }
-
- public _Fields fieldForId(int fieldId) {
- return _Fields.findByThriftId(fieldId);
- }
-
- public void read(org.apache.thrift.protocol.TProtocol iprot) throws org.apache.thrift.TException {
- schemes.get(iprot.getScheme()).getScheme().read(iprot, this);
- }
-
- public void write(org.apache.thrift.protocol.TProtocol oprot) throws org.apache.thrift.TException {
- schemes.get(oprot.getScheme()).getScheme().write(oprot, this);
- }
-
- @Override
- public String toString() {
- StringBuilder sb = new StringBuilder("Log_result(");
- boolean first = true;
-
- sb.append("success:");
- if (this.success == null) {
- sb.append("null");
- } else {
- sb.append(this.success);
- }
- first = false;
- sb.append(")");
- return sb.toString();
- }
-
- public void validate() throws org.apache.thrift.TException {
- // check for required fields
- // check for sub-struct validity
- }
-
- private void writeObject(java.io.ObjectOutputStream out) throws java.io.IOException {
- try {
- write(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(out)));
- } catch (org.apache.thrift.TException te) {
- throw new java.io.IOException(te);
- }
- }
-
- private void readObject(java.io.ObjectInputStream in) throws java.io.IOException, ClassNotFoundException {
- try {
- read(new org.apache.thrift.protocol.TCompactProtocol(new org.apache.thrift.transport.TIOStreamTransport(in)));
- } catch (org.apache.thrift.TException te) {
- throw new java.io.IOException(te);
- }
- }
-
- private static class Log_resultStandardSchemeFactory implements SchemeFactory {
- public Log_resultStandardScheme getScheme() {
- return new Log_resultStandardScheme();
- }
- }
-
- private static class Log_resultStandardScheme extends StandardScheme<Log_result> {
-
- public void read(org.apache.thrift.protocol.TProtocol iprot, Log_result struct) throws org.apache.thrift.TException {
- org.apache.thrift.protocol.TField schemeField;
- iprot.readStructBegin();
- while (true)
- {
- schemeField = iprot.readFieldBegin();
- if (schemeField.type == org.apache.thrift.protocol.TType.STOP) {
- break;
- }
- switch (schemeField.id) {
- case 0: // SUCCESS
- if (schemeField.type == org.apache.thrift.protocol.TType.I32) {
- struct.success = org.apache.flume.source.scribe.ResultCode.findByValue(iprot.readI32());
- struct.setSuccessIsSet(true);
- } else {
- org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
- }
- break;
- default:
- org.apache.thrift.protocol.TProtocolUtil.skip(iprot, schemeField.type);
- }
- iprot.readFieldEnd();
- }
- iprot.readStructEnd();
-
- // check for required fields of primitive type, which can't be checked in the validate method
- struct.validate();
- }
-
- public void write(org.apache.thrift.protocol.TProtocol oprot, Log_result struct) throws org.apache.thrift.TException {
- struct.validate();
-
- oprot.writeStructBegin(STRUCT_DESC);
- if (struct.success != null) {
- oprot.writeFieldBegin(SUCCESS_FIELD_DESC);
- oprot.writeI32(struct.success.getValue());
- oprot.writeFieldEnd();
- }
- oprot.writeFieldStop();
- oprot.writeStructEnd();
- }
-
- }
-
- private static class Log_resultTupleSchemeFactory implements SchemeFactory {
- public Log_resultTupleScheme getScheme() {
- return new Log_resultTupleScheme();
- }
- }
-
- private static class Log_resultTupleScheme extends TupleScheme<Log_result> {
-
- @Override
- public void write(org.apache.thrift.protocol.TProtocol prot, Log_result struct) throws org.apache.thrift.TException {
- TTupleProtocol oprot = (TTupleProtocol) prot;
- BitSet optionals = new BitSet();
- if (struct.isSetSuccess()) {
- optionals.set(0);
- }
- oprot.writeBitSet(optionals, 1);
- if (struct.isSetSuccess()) {
- oprot.writeI32(struct.success.getValue());
- }
- }
-
- @Override
- public void read(org.apache.thrift.protocol.TProtocol prot, Log_result struct) throws org.apache.thrift.TException {
- TTupleProtocol iprot = (TTupleProtocol) prot;
- BitSet incoming = iprot.readBitSet(1);
- if (incoming.get(0)) {
- struct.success = org.apache.flume.source.scribe.ResultCode.findByValue(iprot.readI32());
- struct.setSuccessIsSet(true);
- }
- }
- }
-
- }
-
-}
diff --git a/flume-ng-sources/flume-scribe-source/src/main/java/org/apache/flume/source/scribe/ScribeSource.java b/flume-ng-sources/flume-scribe-source/src/main/java/org/apache/flume/source/scribe/ScribeSource.java
deleted file mode 100644
index e057c98e8..000000000
--- a/flume-ng-sources/flume-scribe-source/src/main/java/org/apache/flume/source/scribe/ScribeSource.java
+++ /dev/null
@@ -1,185 +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.source.scribe;
-
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-
-import org.apache.flume.Context;
-import org.apache.flume.Event;
-import org.apache.flume.EventDrivenSource;
-import org.apache.flume.conf.Configurable;
-import org.apache.flume.event.EventBuilder;
-import org.apache.flume.instrumentation.SourceCounter;
-import org.apache.flume.source.AbstractSource;
-import org.apache.flume.source.scribe.Scribe.Iface;
-import org.apache.thrift.TException;
-import org.apache.thrift.protocol.TBinaryProtocol;
-import org.apache.thrift.server.THsHaServer;
-import org.apache.thrift.server.TServer;
-import org.apache.thrift.transport.TNonblockingServerSocket;
-import org.apache.thrift.transport.TNonblockingServerTransport;
-import org.apache.thrift.transport.layered.TFramedTransport;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * Flume should adopt the Scribe entry {@code LogEntry} from existing
- * Scribe system. Mostly, we may receive message from local Scribe and Flume
- * take responsibility of central Scribe.
- *
- * <p>
- * We use Thrift without deserializing, throughput has 2X increasing
- */
-public class ScribeSource extends AbstractSource implements
- EventDrivenSource, Configurable {
-
- private static final Logger LOG = LoggerFactory.getLogger(ScribeSource.class);
-
- public static final String SCRIBE_CATEGORY = "category";
-
- private static final int DEFAULT_PORT = 1499;
- private static final int DEFAULT_WORKERS = 5;
- private static final int DEFAULT_MAX_READ_BUFFER_BYTES = 16384000;
-
- private TServer server;
- private int port;
- private int workers;
- private int maxReadBufferBytes;
-
- private SourceCounter sourceCounter;
-
- @Override
- public void configure(Context context) {
- port = context.getInteger("port", DEFAULT_PORT);
- maxReadBufferBytes = context.getInteger("maxReadBufferBytes", DEFAULT_MAX_READ_BUFFER_BYTES);
- if(maxReadBufferBytes <= 0){
- maxReadBufferBytes = DEFAULT_MAX_READ_BUFFER_BYTES;
- }
-
- workers = context.getInteger("workerThreads", DEFAULT_WORKERS);
- if (workers <= 0) {
- workers = DEFAULT_WORKERS;
- }
-
- if (sourceCounter == null) {
- sourceCounter = new SourceCounter(getName());
- }
- }
-
- private class Startup extends Thread {
-
- public void run() {
- try {
- Scribe.Processor processor = new Scribe.Processor(new Receiver());
- TNonblockingServerTransport transport = new TNonblockingServerSocket(port);
- THsHaServer.Args args = new THsHaServer.Args(transport);
-
- args.minWorkerThreads(workers);
- args.maxWorkerThreads(workers);
- args.processor(processor);
- args.transportFactory(new TFramedTransport.Factory(maxReadBufferBytes));
- args.protocolFactory(new TBinaryProtocol.Factory(false, false));
- args.maxReadBufferBytes = maxReadBufferBytes;
-
- server = new THsHaServer(args);
-
- LOG.info("Starting Scribe Source on port " + port);
-
- server.serve();
- } catch (Exception e) {
- LOG.warn("Scribe failed", e);
- }
- }
-
- }
-
- @Override
- public void start() {
- Startup startupThread = new Startup();
- startupThread.start();
-
- try {
- Thread.sleep(3000);
- } catch (InterruptedException e) {}
-
- if (!server.isServing()) {
- throw new IllegalStateException("Failed initialization of ScribeSource");
- }
-
- sourceCounter.start();
- super.start();
- }
-
- @Override
- public void stop() {
- LOG.info("Scribe source stopping");
-
- if (server != null) {
- server.stop();
- }
-
- sourceCounter.stop();
- super.stop();
-
- LOG.info("Scribe source stopped. Metrics:{}", sourceCounter);
- }
-
-
- class Receiver implements Iface {
-
- public ResultCode Log(List<LogEntry> list) throws TException {
- if (list != null) {
- sourceCounter.addToEventReceivedCount(list.size());
-
- try {
- List<Event> events = new ArrayList<Event>(list.size());
-
- for (LogEntry entry : list) {
- Map<String, String> headers = new HashMap<String, String>(1, 1);
- String category = entry.getCategory();
-
- if (category != null) {
- headers.put(SCRIBE_CATEGORY, category);
- }
-
- Event event = EventBuilder.withBody(entry.getMessage().getBytes(), headers);
- events.add(event);
- }
-
- if (events.size() > 0) {
- getChannelProcessor().processEventBatch(events);
- }
-
- sourceCounter.addToEventAcceptedCount(list.size());
- return ResultCode.OK;
- } catch (Exception e) {
- LOG.warn("Scribe source handling failure", e);
- sourceCounter.incrementEventReadOrChannelFail(e);
- }
- }
-
- return ResultCode.TRY_LATER;
- }
- }
-
-}
diff --git a/flume-ng-sources/flume-scribe-source/src/main/thrift/aslv2 b/flume-ng-sources/flume-scribe-source/src/main/thrift/aslv2
deleted file mode 100644
index 7243427b1..000000000
--- a/flume-ng-sources/flume-scribe-source/src/main/thrift/aslv2
+++ /dev/null
@@ -1,18 +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.
- */
diff --git a/flume-ng-sources/flume-scribe-source/src/main/thrift/scribe-source.thrift b/flume-ng-sources/flume-scribe-source/src/main/thrift/scribe-source.thrift
deleted file mode 100644
index 6594ee4e7..000000000
--- a/flume-ng-sources/flume-scribe-source/src/main/thrift/scribe-source.thrift
+++ /dev/null
@@ -1,34 +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.
- */
-
-
-namespace java org.apache.flume.source.scribe
-
-enum ResultCode {
- OK,TRY_LATER
-}
-
-struct LogEntry {
- 1: string category,
- 2: string message
-}
-
-service Scribe {
- ResultCode Log(1: list<LogEntry> messages);
-}
\ No newline at end of file
diff --git a/flume-ng-sources/flume-scribe-source/src/test/java/org/apache/flume/source/scribe/TestScribeSource.java b/flume-ng-sources/flume-scribe-source/src/test/java/org/apache/flume/source/scribe/TestScribeSource.java
deleted file mode 100644
index 7cf101ebd..000000000
--- a/flume-ng-sources/flume-scribe-source/src/test/java/org/apache/flume/source/scribe/TestScribeSource.java
+++ /dev/null
@@ -1,165 +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.source.scribe;
-
-import junit.framework.Assert;
-import org.apache.flume.*;
-import org.apache.flume.channel.ChannelProcessor;
-import org.apache.flume.channel.MemoryChannel;
-import org.apache.flume.channel.ReplicatingChannelSelector;
-import org.apache.flume.conf.Configurables;
-import org.apache.flume.instrumentation.SourceCounter;
-import org.apache.thrift.protocol.TBinaryProtocol;
-import org.apache.thrift.protocol.TProtocol;
-import org.apache.thrift.transport.TSocket;
-import org.apache.thrift.transport.TTransport;
-import org.apache.thrift.transport.layered.TFramedTransport;
-import org.junit.AfterClass;
-import org.junit.BeforeClass;
-import org.junit.Test;
-import org.mockito.internal.util.reflection.Whitebox;
-
-import java.io.IOException;
-import java.net.ServerSocket;
-import java.util.ArrayList;
-import java.util.List;
-
-import static org.mockito.Matchers.anyListOf;
-import static org.mockito.Mockito.doThrow;
-import static org.mockito.Mockito.mock;
-
-/**
- *
- */
-public class TestScribeSource {
-
- private static int port;
- private static Channel memoryChannel;
- private static ScribeSource scribeSource;
-
- private static int findFreePort() throws IOException {
- ServerSocket socket = new ServerSocket(0);
- int port = socket.getLocalPort();
- socket.close();
- return port;
- }
-
- @BeforeClass
- public static void setUpClass() throws Exception {
- port = findFreePort();
- Context context = new Context();
- context.put("port", String.valueOf(port));
-
- scribeSource = new ScribeSource();
- scribeSource.setName("Scribe Source");
-
- Configurables.configure(scribeSource, context);
-
- memoryChannel = new MemoryChannel();
- Configurables.configure(memoryChannel, context);
-
- List<Channel> channels = new ArrayList<Channel>(1);
- channels.add(memoryChannel);
-
- ChannelSelector rcs = new ReplicatingChannelSelector();
- rcs.setChannels(channels);
-
- memoryChannel.start();
-
- scribeSource.setChannelProcessor(new ChannelProcessor(rcs));
- scribeSource.start();
- }
-
- private void sendSingle() throws org.apache.thrift.TException {
- TTransport transport = new TFramedTransport(new TSocket("localhost", port));
-
- TProtocol protocol = new TBinaryProtocol(transport);
- Scribe.Client client = new Scribe.Client(protocol);
- transport.open();
- LogEntry logEntry = new LogEntry("INFO", "Sending info msg to scribe source");
- List<LogEntry> logEntries = new ArrayList<LogEntry>(1);
- logEntries.add(logEntry);
- client.Log(logEntries);
- }
-
- @Test
- public void testScribeMessage() throws Exception {
- sendSingle();
-
- // try to get it from Channels
- Transaction tx = memoryChannel.getTransaction();
- tx.begin();
- Event e = memoryChannel.take();
- Assert.assertNotNull(e);
- Assert.assertEquals("Sending info msg to scribe source", new String(e.getBody()));
- tx.commit();
- tx.close();
- }
-
- @Test
- public void testScribeMultipleMessages() throws Exception {
- TTransport transport = new TFramedTransport(new TSocket("localhost", port));
-
- TProtocol protocol = new TBinaryProtocol(transport);
- Scribe.Client client = new Scribe.Client(protocol);
- transport.open();
-
- List<LogEntry> logEntries = new ArrayList<LogEntry>(10);
- for (int i = 0; i < 10; i++) {
- LogEntry logEntry = new LogEntry("INFO", String.format("Sending info msg# %d to scribe source", i));
- logEntries.add(logEntry);
- }
-
- client.Log(logEntries);
-
- // try to get it from Channels
- Transaction tx = memoryChannel.getTransaction();
- tx.begin();
-
- for (int i = 0; i < 10; i++) {
- Event e = memoryChannel.take();
- Assert.assertNotNull(e);
- Assert.assertEquals(String.format("Sending info msg# %d to scribe source", i), new String(e.getBody()));
- }
- tx.commit();
- tx.close();
- }
-
- @Test
- public void testErrorCounter() throws Exception {
- ChannelProcessor cp = mock(ChannelProcessor.class);
- doThrow(new ChannelException("dummy")).when(cp).processEventBatch(anyListOf(Event.class));
- ChannelProcessor origCp = scribeSource.getChannelProcessor();
- scribeSource.setChannelProcessor(cp);
-
- sendSingle();
-
- scribeSource.setChannelProcessor(origCp);
-
- SourceCounter sc = (SourceCounter) Whitebox.getInternalState(scribeSource, "sourceCounter");
- org.junit.Assert.assertEquals(1, sc.getChannelWriteFail());
- }
-
- @AfterClass
- public static void cleanup() {
- memoryChannel.stop();
- scribeSource.stop();
- }
-
-}
diff --git a/flume-ng-sources/pom.xml b/flume-ng-sources/pom.xml
index d0dcbc4f3..5280a3ca0 100644
--- a/flume-ng-sources/pom.xml
+++ b/flume-ng-sources/pom.xml
@@ -33,7 +33,6 @@ limitations under the License.
<packaging>pom</packaging>
<modules>
- <module>flume-jms-source</module>
<module>flume-taildir-source</module>
</modules>