[FLINK-23322][connectors/rabbitmq] Increase RMQSource handshake timeo…

…ut tolerating network congestions
apache · Jul 29, 2021 · 4e14b3f · 4e14b3f
1 parent be67258
commit 4e14b3f
Show file tree

Hide file tree

Showing 2 changed files with 37 additions and 0 deletions.
diff --git a/...abbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSourceITCase.java b/...abbitmq/src/test/java/org/apache/flink/streaming/connectors/rabbitmq/RMQSourceITCase.java
@@ -42,7 +42,10 @@
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.testcontainers.containers.RabbitMQContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
 import org.testcontainers.containers.wait.strategy.Wait;
 import org.testcontainers.utility.DockerImageName;
 
@@ -53,6 +56,10 @@
 /** A class containing RabbitMQ source tests against a real RabbiMQ cluster. */
 public class RMQSourceITCase {
 
+    private static final Logger LOG = LoggerFactory.getLogger(RMQSourceITCase.class);
+    private static final Slf4jLogConsumer LOG_CONSUMER = new Slf4jLogConsumer(LOG);
+
+    private static final int HANDSHAKE_TIMEOUT = 30000;
     private static final int RABBITMQ_PORT = 5672;
     private static final String QUEUE_NAME = "test-queue";
     private static final JobID JOB_ID = new JobID();
@@ -75,6 +82,7 @@ public class RMQSourceITCase {
             new RabbitMQContainer(
                             DockerImageName.parse("rabbitmq").withTag("3.7.25-management-alpine"))
                     .withExposedPorts(RABBITMQ_PORT)
+                    .withLogConsumer(LOG_CONSUMER)
                     .waitingFor(Wait.forListeningPort());
 
     @Before
@@ -127,6 +135,7 @@ private static Connection getRMQConnection() throws IOException, TimeoutExceptio
         ConnectionFactory factory = new ConnectionFactory();
         factory.setUsername(RMQ_CONTAINER.getAdminUsername());
         factory.setPassword(RMQ_CONTAINER.getAdminPassword());
+        factory.setHandshakeTimeout(HANDSHAKE_TIMEOUT);
         factory.setVirtualHost("/");
         factory.setHost(RMQ_CONTAINER.getHost());
         factory.setPort(RMQ_CONTAINER.getAmqpPort());

diff --git a/flink-connectors/flink-connector-rabbitmq/src/test/resources/log4j2-test.properties b/flink-connectors/flink-connector-rabbitmq/src/test/resources/log4j2-test.properties
@@ -0,0 +1,28 @@
+################################################################################
+#  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.
+################################################################################
+
+# Set root logger level to OFF to not flood build logs
+# set manually to INFO for debugging purposes
+rootLogger.level = OFF
+rootLogger.appenderRef.test.ref = TestLogger
+
+appender.testlogger.name = TestLogger
+appender.testlogger.type = CONSOLE
+appender.testlogger.target = SYSTEM_ERR
+appender.testlogger.layout.type = PatternLayout
+appender.testlogger.layout.pattern = %-4r [%t] %-5p %c %x - %m%n