You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ec...@apache.org on 2025/09/22 09:15:24 UTC
(flink-connector-cassandra) 06/08: [FLINK-37937] Improve start/stop for the 2-container cluster.
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-cassandra.git
commit e53d28a6e14c53a3b7f0556b4f1c493129327e42
Author: Etienne Chauchot <ec...@apache.org>
AuthorDate: Mon Sep 8 11:32:40 2025 +0200
[FLINK-37937] Improve start/stop for the 2-container cluster.
---
.../cassandra/CassandraTestEnvironment.java | 28 ++++++++++++++--------
1 file changed, 18 insertions(+), 10 deletions(-)
diff --git a/flink-connector-cassandra/src/test/java/org/apache/flink/connector/cassandra/CassandraTestEnvironment.java b/flink-connector-cassandra/src/test/java/org/apache/flink/connector/cassandra/CassandraTestEnvironment.java
index 24e8fda..bb8d2e5 100644
--- a/flink-connector-cassandra/src/test/java/org/apache/flink/connector/cassandra/CassandraTestEnvironment.java
+++ b/flink-connector-cassandra/src/test/java/org/apache/flink/connector/cassandra/CassandraTestEnvironment.java
@@ -37,9 +37,10 @@ import org.testcontainers.containers.Container.ExecResult;
import org.testcontainers.containers.Network;
import org.testcontainers.containers.output.OutputFrame;
import org.testcontainers.containers.output.Slf4jLogConsumer;
-import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.containers.wait.CassandraQueryWaitStrategy;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.lifecycle.Startables;
import org.testcontainers.utility.MountableFile;
import java.net.InetSocketAddress;
@@ -143,19 +144,16 @@ public class CassandraTestEnvironment implements TestResource {
// configure container start to wait until cassandra is ready to receive queries
// start with retrials
cassandraContainer1.waitingFor(
- Wait.forLogMessage(".*Startup complete.*", 1)
- .withStartupTimeout(Duration.ofMinutes(2)));
- cassandraContainer1.start();
+ new CassandraQueryWaitStrategy().withStartupTimeout(Duration.ofMinutes(2)));
+ cassandraContainer2.waitingFor(
+ new CassandraQueryWaitStrategy().withStartupTimeout(Duration.ofMinutes(2)));
+ Startables.deepStart(cassandraContainer1, cassandraContainer2).join();
cassandraContainer1.followOutput(
new Slf4jLogConsumer(LOG),
OutputFrame.OutputType.END,
OutputFrame.OutputType.STDERR,
OutputFrame.OutputType.STDOUT);
- cassandraContainer2.waitingFor(
- Wait.forLogMessage(".*Startup complete.*", 1)
- .withStartupTimeout(Duration.ofMinutes(2)));
- cassandraContainer2.start();
cassandraContainer2.followOutput(
new Slf4jLogConsumer(LOG),
OutputFrame.OutputType.END,
@@ -201,8 +199,18 @@ public class CassandraTestEnvironment implements TestResource {
if (cluster != null) {
cluster.close();
}
- cassandraContainer1.stop();
- cassandraContainer2.stop();
+ try {
+ cassandraContainer1.stop();
+ } catch (Exception e) {
+ // do not fail the test for a stop failure and allow the other container to stop
+ LOG.error("Cassandra test container 1 failed to stop.", e);
+ }
+ try {
+ cassandraContainer2.stop();
+ } catch (Exception e) {
+ // do not fail the test for a stop failure
+ LOG.error("Cassandra test container 2 failed to stop.", e);
+ }
}
private ClusterBuilder createBuilderWithConsistencyLevel(