diff --git a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/it/SolaceContainerManager.java b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/it/SolaceContainerManager.java index d83a63c743cf..c029cc632b39 100644 --- a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/it/SolaceContainerManager.java +++ b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/it/SolaceContainerManager.java @@ -18,7 +18,6 @@ package org.apache.beam.sdk.io.solace.it; import java.io.IOException; -import java.net.ServerSocket; import java.util.ArrayList; import java.util.Arrays; import java.util.List; @@ -37,25 +36,29 @@ public class SolaceContainerManager { public static final String USERNAME = "username"; public static final String TOPIC_NAME = "test_topic"; private static final Logger LOG = LoggerFactory.getLogger(SolaceContainerManager.class); + private static final int JCSMP_PORT = 55555; + private static final int SEMP_PORT = 8080; + private static final int MAX_COMMAND_ATTEMPTS = 30; private final SolaceContainer container; - int jcsmpPortMapped = findAvailablePort(); - int sempPortMapped = findAvailablePort(); + int jcsmpPortMapped; + int sempPortMapped; - public SolaceContainerManager() throws IOException { + public SolaceContainerManager() { + // Let Docker pick the host ports. Reserving a free port up front and binding it later races + // with anything else that grabs the port in between. this.container = - new SolaceContainer(DockerImageName.parse("solace/solace-pubsub-standard:10.7")) { - { - addFixedExposedPort(jcsmpPortMapped, 55555); - addFixedExposedPort(sempPortMapped, 8080); - } - }.withVpn(VPN_NAME) + new SolaceContainer(DockerImageName.parse("solace/solace-pubsub-standard:10.7")) + .withVpn(VPN_NAME) .withCredentials(USERNAME, PASSWORD) .withTopic(TOPIC_NAME, Service.SMF) + .withExposedPorts(SEMP_PORT, JCSMP_PORT) .withLogConsumer(new Slf4jLogConsumer(LOG)); } public void start() { container.start(); + jcsmpPortMapped = container.getMappedPort(JCSMP_PORT); + sempPortMapped = container.getMappedPort(SEMP_PORT); } void createQueueWithSubscriptionTopic(String queueName) { @@ -79,8 +82,11 @@ void createQueueWithSubscriptionTopic(String queueName) { "{\"topicEndpointName\":\"" + TOPIC_NAME + "\",\"accessType\":\"exclusive\",\"permission\":\"modify-topic\",\"ingressEnabled\":true,\"egressEnabled\":true}"); - executeCommand( + // The tests read from this queue, so fail the setup instead of letting them time out later. + // --fail makes curl exit with an error on HTTP errors, e.g. while SEMP is still starting up. + executeCommandWithRetries( "curl", + "--fail", "http://localhost:8080/SEMP/v2/config/msgVpns/" + VPN_NAME + "/queues", "-X", "POST", @@ -92,8 +98,9 @@ void createQueueWithSubscriptionTopic(String queueName) { "{\"queueName\":\"" + queueName + "\",\"accessType\":\"non-exclusive\",\"maxMsgSpoolUsage\":200,\"permission\":\"consume\",\"ingressEnabled\":true,\"egressEnabled\":true}"); - executeCommand( + executeCommandWithRetries( "curl", + "--fail", "http://localhost:8080/SEMP/v2/config/msgVpns/" + VPN_NAME + "/queues/" @@ -109,17 +116,35 @@ void createQueueWithSubscriptionTopic(String queueName) { "{\"subscriptionTopic\":\"" + TOPIC_NAME + "\"}"); } - private void executeCommand(String... command) { + private boolean executeCommand(String... command) { try { ExecResult execResult = container.execInContainer(command); if (execResult.getExitCode() != 0) { logCommandError(execResult.getStderr(), command); - } else { - LOG.info("{}", execResult.getStdout()); + return false; } + LOG.info("{}", execResult.getStdout()); + return true; } catch (IOException | InterruptedException e) { logCommandError(e.getMessage(), command); + return false; + } + } + + private void executeCommandWithRetries(String... command) { + for (int attempt = 1; attempt <= MAX_COMMAND_ATTEMPTS; attempt++) { + if (executeCommand(command)) { + return; + } + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } } + throw new IllegalStateException( + "Could not execute command " + String.join(" ", command) + " in the Solace container"); } private void logCommandError(String error, String... command) { @@ -170,19 +195,4 @@ public void sendToTopic(String payload, List additionalHeaders) { executeCommand(command.toArray(new String[0])); } - - private static int findAvailablePort() throws IOException { - ServerSocket s = new ServerSocket(0); - try { - return s.getLocalPort(); - } finally { - s.close(); - try { - // Some systems don't free the port for future use immediately. - Thread.sleep(100); - } catch (InterruptedException exn) { - // ignore - } - } - } } diff --git a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/it/SolaceIOCustomSessionServiceFactoryIT.java b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/it/SolaceIOCustomSessionServiceFactoryIT.java index b047026a2bbd..e92670caa8da 100644 --- a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/it/SolaceIOCustomSessionServiceFactoryIT.java +++ b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/it/SolaceIOCustomSessionServiceFactoryIT.java @@ -42,7 +42,6 @@ import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.testing.TestPipelineOptions; import org.apache.beam.sdk.testing.TestStream; -import org.apache.beam.sdk.testutils.metrics.MetricsReader; import org.apache.beam.sdk.transforms.DoFn; import org.apache.beam.sdk.transforms.MapElements; import org.apache.beam.sdk.transforms.ParDo; @@ -61,6 +60,7 @@ public class SolaceIOCustomSessionServiceFactoryIT { private static final String READ_COUNT = "read_count"; private static final String QUEUE_NAME = "test_queue"; private static final long PUBLISH_MESSAGE_COUNT = 20; + private static final Duration READ_TIMEOUT = Duration.standardMinutes(2); private static final TestPipelineOptions pipelineOptions; private static SolaceContainerManager solaceContainerManager; @@ -94,7 +94,7 @@ public static void afterClass() { * that with the custom {@link FixedCredentialsBasicAuthJcsmpSessionServiceFactory}. */ @Test - public void test01writeAndReadWithCustomSessionServiceFactory() { + public void test01writeAndReadWithCustomSessionServiceFactory() throws IOException { Pipeline writerPipeline = createWriterPipeline(WriterType.BATCHED); writerPipeline .apply( @@ -116,12 +116,9 @@ public void test01writeAndReadWithCustomSessionServiceFactory() { .apply("Count", ParDo.of(new CountingFn<>(NAMESPACE, READ_COUNT))); PipelineResult pipelineResult = writerPipeline.run(); - // We need enough time for Beam to pull all messages from the queue, but we need a timeout too, - // as the Read connector will keep attempting to read forever. - pipelineResult.waitUntilFinish(Duration.standardSeconds(15)); - - MetricsReader metricsReader = new MetricsReader(pipelineResult, NAMESPACE); - long actualRecordsCount = metricsReader.getCounterMetric(READ_COUNT); + long actualRecordsCount = + SolaceTestUtils.waitForCounterAndCancel( + pipelineResult, NAMESPACE, READ_COUNT, PUBLISH_MESSAGE_COUNT, READ_TIMEOUT); assertEquals(PUBLISH_MESSAGE_COUNT, actualRecordsCount); } diff --git a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/it/SolaceIOIT.java b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/it/SolaceIOIT.java index 8311a67bf6c1..8076ac816108 100644 --- a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/it/SolaceIOIT.java +++ b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/it/SolaceIOIT.java @@ -66,6 +66,7 @@ public class SolaceIOIT { private static final String queueName = "test_queue"; private static final TestPipelineOptions pipelineOptions; private static final long PUBLISH_MESSAGE_COUNT = 20; + private static final Duration READ_TIMEOUT = Duration.standardMinutes(2); static { pipelineOptions = PipelineOptionsFactory.create().as(TestPipelineOptions.class); @@ -100,7 +101,7 @@ public void test01WriteStreaming() { } @Test - public void test02Read() { + public void test02Read() throws IOException { pipeline .apply( "Read from Solace", @@ -126,12 +127,9 @@ public void test02Read() { .apply("Count", ParDo.of(new CountingFn<>(NAMESPACE, READ_COUNT))); PipelineResult pipelineResult = pipeline.run(); - // We need enough time for Beam to pull all messages from the queue, but we need a timeout too, - // as the Read connector will keep attempting to read forever. - pipelineResult.waitUntilFinish(Duration.standardSeconds(15)); - - MetricsReader metricsReader = new MetricsReader(pipelineResult, NAMESPACE); - long actualRecordsCount = metricsReader.getCounterMetric(READ_COUNT); + long actualRecordsCount = + SolaceTestUtils.waitForCounterAndCancel( + pipelineResult, NAMESPACE, READ_COUNT, PUBLISH_MESSAGE_COUNT, READ_TIMEOUT); assertEquals(PUBLISH_MESSAGE_COUNT, actualRecordsCount); } diff --git a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/it/SolaceIOMultipleSempIT.java b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/it/SolaceIOMultipleSempIT.java index 77d00b4e41ec..18c2d30a2a63 100644 --- a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/it/SolaceIOMultipleSempIT.java +++ b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/it/SolaceIOMultipleSempIT.java @@ -43,7 +43,6 @@ import org.apache.beam.sdk.testing.TestPipeline; import org.apache.beam.sdk.testing.TestPipelineOptions; import org.apache.beam.sdk.testing.TestStream; -import org.apache.beam.sdk.testutils.metrics.MetricsReader; import org.apache.beam.sdk.transforms.DoFn; import org.apache.beam.sdk.transforms.MapElements; import org.apache.beam.sdk.transforms.ParDo; @@ -62,6 +61,7 @@ public class SolaceIOMultipleSempIT { private static final String READ_COUNT = "read_count"; private static final String QUEUE_NAME = "test_queue"; private static final long PUBLISH_MESSAGE_COUNT = 20; + private static final Duration READ_TIMEOUT = Duration.standardMinutes(2); private static final TestPipelineOptions pipelineOptions; private static SolaceContainerManager solaceContainerManager; @@ -98,7 +98,7 @@ public static void afterClass() { * with the Solace message broker. */ @Test - public void test01writeAndReadWithMultipleSempClientFactory() { + public void test01writeAndReadWithMultipleSempClientFactory() throws IOException { Pipeline writerPipeline = createWriterPipeline(WriterType.BATCHED, solaceContainerManager.jcsmpPortMapped); writerPipeline @@ -129,12 +129,9 @@ public void test01writeAndReadWithMultipleSempClientFactory() { .apply("Count", ParDo.of(new CountingFn<>(NAMESPACE, READ_COUNT))); PipelineResult pipelineResult = writerPipeline.run(); - // We need enough time for Beam to pull all messages from the queue, but we need a timeout too, - // as the Read connector will keep attempting to read forever. - pipelineResult.waitUntilFinish(Duration.standardSeconds(15)); - - MetricsReader metricsReader = new MetricsReader(pipelineResult, NAMESPACE); - long actualRecordsCount = metricsReader.getCounterMetric(READ_COUNT); + long actualRecordsCount = + SolaceTestUtils.waitForCounterAndCancel( + pipelineResult, NAMESPACE, READ_COUNT, PUBLISH_MESSAGE_COUNT, READ_TIMEOUT); assertEquals(PUBLISH_MESSAGE_COUNT, actualRecordsCount); } diff --git a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/it/SolaceTestUtils.java b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/it/SolaceTestUtils.java new file mode 100644 index 000000000000..9d5717e74f95 --- /dev/null +++ b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/it/SolaceTestUtils.java @@ -0,0 +1,57 @@ +/* + * 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.beam.sdk.io.solace.it; + +import java.io.IOException; +import org.apache.beam.sdk.PipelineResult; +import org.apache.beam.sdk.testutils.metrics.MetricsReader; +import org.joda.time.Duration; +import org.joda.time.Instant; + +class SolaceTestUtils { + private static final Duration POLL_INTERVAL = Duration.standardSeconds(1); + + /** + * Waits until the counter reaches {@code expectedCount} or {@code timeout} passes, then cancels + * the pipeline and returns the last value of the counter. + * + *

The Read connector keeps polling the queue forever, so the pipeline never finishes on its + * own. Waiting for the expected count instead of a fixed amount of time keeps the tests from + * stopping too early on a slow machine. + */ + static long waitForCounterAndCancel( + PipelineResult pipelineResult, + String namespace, + String counterName, + long expectedCount, + Duration timeout) + throws IOException { + MetricsReader metricsReader = new MetricsReader(pipelineResult, namespace); + Instant deadline = Instant.now().plus(timeout); + long count = metricsReader.getCounterMetric(counterName); + while (count < expectedCount && Instant.now().isBefore(deadline)) { + PipelineResult.State state = pipelineResult.waitUntilFinish(POLL_INTERVAL); + count = metricsReader.getCounterMetric(counterName); + if (state != null && state.isTerminal()) { + break; + } + } + pipelineResult.cancel(); + return count; + } +}