Merge pull request #5 from zhipcui/master

support handshakeTimeout option

pom.xml

@@ -27,8 +27,8 @@
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
-        <flume.version>1.4.0</flume.version>
-        <rabbitmq.version>3.3.5</rabbitmq.version>
+        <flume.version>1.8.0</flume.version>
+        <rabbitmq.version>3.6.6</rabbitmq.version>
         <junit.version>4.10</junit.version>
         <slf4j-api.version>1.7.2</slf4j-api.version>
         <mockito-all.version>1.9.5</mockito-all.version>

src/main/java/org/apache/flume/amqp/AmqpConsumer.java

@@ -30,6 +30,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.concurrent.TimeoutException;
 
 import static com.google.common.base.Preconditions.checkArgument;
 
@@ -286,7 +287,7 @@ protected String declarationsForChannel(Channel channel) throws IOException {
         return queueName;
     }
 
-    private Connection createConnection() throws IOException {
+    private Connection createConnection() throws IOException, TimeoutException {
         LOG.info("Connecting to {} ...", connectionFactory.getHost());
         Connection conn = connectionFactory.newConnection();
         LOG.info("Connected to {}", connectionFactory.getHost());

src/main/java/org/apache/flume/amqp/AmqpSource.java

@@ -150,6 +150,7 @@ static ConnectionFactory createConnectionFactoryFrom(Context context) {
         String password = context.getString(AmqpSourceConfigurationConstants.PASSWORD, Constants.Defaults.PASSWORD);
         int connectionTimeout = context.getInteger(AmqpSourceConfigurationConstants.CONNECTION_TIMEOUT, Constants.Defaults.CONNECTION_TIMEOUT);
         int requestHeartbeat = context.getInteger(AmqpSourceConfigurationConstants.REQUEST_HEARTBEAT, Constants.Defaults.REQUESTED_HEARTBEAT);
+        int handshakeTimeout = context.getInteger(AmqpSourceConfigurationConstants.HANDSHAKE_TIMEOUT, Constants.Defaults.HANDSHAKE_TIMEOUT);
 
         ConnectionFactory connectionFactory = new ConnectionFactory();
         connectionFactory.setHost(host);
@@ -159,6 +160,7 @@ static ConnectionFactory createConnectionFactoryFrom(Context context) {
         connectionFactory.setPassword(password);
         connectionFactory.setConnectionTimeout(connectionTimeout);
         connectionFactory.setRequestedHeartbeat(requestHeartbeat);
+        connectionFactory.setHandshakeTimeout(handshakeTimeout);
 
         return connectionFactory;
     }

src/main/java/org/apache/flume/amqp/AmqpSourceConfigurationConstants.java

@@ -127,6 +127,12 @@ class AmqpSourceConfigurationConstants {
      */
     public static final String REQUEST_HEARTBEAT = "requestedHeartbeat";
 
+    /**
+     * The default AMQP 0-9-1 connection handshake timeout for TCP (socket) connection timeout. Defaults to
+     * {@link com.rabbitmq.client.ConnectionFactory#DEFAULT_HANDSHAKE_TIMEOUT}
+     */
+    public static final String HANDSHAKE_TIMEOUT = "handshakeTimeout";
+
     /**
      * This property has dual purposes. If the source is not in auto-ack mode, then this will be the number of messages
      * to buffer before sending an ack to the server.

src/main/java/org/apache/flume/amqp/Constants.java

@@ -54,6 +54,7 @@ private Defaults() {
         public static final String PASSWORD = ConnectionFactory.DEFAULT_PASS;
         public static final int CONNECTION_TIMEOUT = ConnectionFactory.DEFAULT_CONNECTION_TIMEOUT;
         public static final int REQUESTED_HEARTBEAT = ConnectionFactory.DEFAULT_HEARTBEAT;
+        public static final int HANDSHAKE_TIMEOUT = ConnectionFactory.DEFAULT_HANDSHAKE_TIMEOUT;
 
     }