github-actions[bot] commented on code in PR #40740: URL: https://github.com/apache/beam/pull/40740#discussion_r4354895487
########## 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. + * + * <p>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(); Review Comment: **🟡 Medium**: If an exception is thrown inside the `while` loop (e.g., from `waitUntilFinish` or `getCounterMetric`), `pipelineResult.cancel()` will not be called, leaving the unbounded streaming pipeline running in the JVM. Consider wrapping the polling loop in a `try ... finally` block so the pipeline is always cancelled if it hasn't reached a terminal state. ########## sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/it/SolaceContainerManager.java: ########## @@ -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) Review Comment: **🟢 Low**: `GenericContainer.withExposedPorts(...)` overwrites the container's exposed ports list rather than appending to it, which replaces the port added by `.withTopic(TOPIC_NAME, Service.SMF)` on the line above. While it works here because `JCSMP_PORT` (55555) is the same as `Service.SMF.getPort()`, `SolaceContainer` already exposes port 8080 (`SEMP_PORT`) in its constructor and `withTopic(..., Service.SMF)` already calls `addExposedPort(55555)`, so `.withExposedPorts(SEMP_PORT, JCSMP_PORT)` is redundant (or `addExposedPorts` could be used if you want to be explicit without overwriting). -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
