Skip to content

Add retry to test connections #23757

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
johnjcasey merged 6 commits into from
Nov 22, 2022
Merged

Add retry to test connections #23757

Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
b250823
Move connection setup logic for JDBCIO WriteFn to @startBundle to lim…
johnjcasey Oct 20, 2022
3eb06bd
move connection setup logic back to processElement.
johnjcasey Oct 24, 2022
d0c38cb
run spotless
johnjcasey Oct 25, 2022
d31b84a
use fluent backoff instead of manual implementation
johnjcasey Oct 26, 2022
d7526fa
refactor to manual resource management
johnjcasey Oct 28, 2022
802601d
run spotless
johnjcasey Oct 28, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 39 additions & 3 deletions sdks/java/io/common/src/test/java/org/apache/beam/sdk/io/common/DatabaseTestHelper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,8 +34,13 @@
import java.util.Optional;
import java.util.stream.Collectors;
import javax.sql.DataSource;
import org.apache.beam.sdk.util.BackOff;
import org.apache.beam.sdk.util.BackOffUtils;
import org.apache.beam.sdk.util.FluentBackoff;
import org.apache.beam.sdk.util.Sleeper;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.Lists;
import org.joda.time.Duration;
import org.postgresql.ds.PGSimpleDataSource;
import org.testcontainers.containers.JdbcDatabaseContainer;

Expand Down Expand Up @@ -86,9 +91,40 @@ public static void createTable(
fieldsAndTypes.stream()
.map(kv -> kv.getKey() + " " + kv.getValue())
.collect(Collectors.joining(", "));
try (Connection connection = dataSource.getConnection()) {
try (Statement statement = connection.createStatement()) {
statement.execute(String.format("create table %s (%s)", tableName, fieldsList));
SQLException exception = null;
Sleeper sleeper = Sleeper.DEFAULT;
BackOff backoff =
FluentBackoff.DEFAULT
.withInitialBackoff(Duration.standardSeconds(1))
.withMaxCumulativeBackoff(Duration.standardMinutes(5))
.withMaxRetries(4)
.backoff();
while (true) {
// This is not implemented as try-with-resources because it appears that try-with-resources is
// not correctly catching the PSQLException thrown by dataSource.getConnection()
Connection connection = null;
try {
connection = dataSource.getConnection();
try (Statement statement = connection.createStatement()) {
statement.execute(String.format("create table %s (%s)", tableName, fieldsList));
return;
}
} catch (SQLException e) {
exception = e;
} finally {
if (connection != null) {
connection.close();
Copy link
Contributor

@Abacn Abacn Oct 28, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Probably it is closing the connection caused issue. dataSource owns a connection pool and manage the connections. Closing it pre-maturely may have caused racing conditions.

Copy link
Contributor Author

@johnjcasey johnjcasey Oct 28, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

it looks like it is happening in .getConnection based on the stacktrace

}
}
boolean hasNext;
try {
hasNext = BackOffUtils.next(sleeper, backoff);
} catch (Exception e) {
throw new RuntimeException(e);
}
if (!hasNext) {
// we tried the max number of times
throw exception;
}
}
}
Expand Down