bvolpato commented on code in PR #40298:
URL: https://github.com/apache/beam/pull/40298#discussion_r4369797950
##########
sdks/java/io/influxdb/src/test/java/org/apache/beam/sdk/io/influxdb/InfluxDbIOIT.java:
##########
@@ -137,7 +142,7 @@ public void createRetentionPolicyInDB(
connection.query(
new Query(
String.format(
- "CREATE RETENTION POLICY %s ON %s DURATION 1d REPLICATION 1",
+ "ALTER RETENTION POLICY %s ON %s DURATION 1d REPLICATION 1",
Review Comment:
Could we keep `CREATE` for the custom retention policies? Each test starts
with a fresh database, so `ALTER` fails because those policies don’t exist.
##########
runners/prism/java/src/main/java/org/apache/beam/runners/prism/PrismRunner.java:
##########
@@ -104,18 +123,52 @@ PrismExecutor startPrism() throws IOException {
String logLevelFlag =
String.format(
PrismExecutor.LOG_LEVEL_FLAG_TEMPLATE,
prismPipelineOptions.getPrismLogLevel());
- String endpoint = "localhost:" + port;
- prismPipelineOptions.setJobEndpoint(endpoint);
- String command = locator.resolve();
- PrismExecutor executor =
- PrismExecutor.builder()
- .setCommand(command)
- .setArguments(
- Arrays.asList(portFlag, serveHttpFlag,
idleShutdownTimeoutFlag, logLevelFlag))
- .build();
- executor.execute();
- checkState(executor.isAlive());
- return executor;
+
+ for (int attempt = 1; ; attempt++) {
+ int port = findAvailablePort();
+ String portFlag = String.format(PrismExecutor.JOB_PORT_FLAG_TEMPLATE,
port);
+ String endpoint = "localhost:" + port;
+ prismPipelineOptions.setJobEndpoint(endpoint);
+ PrismExecutor executor =
+ PrismExecutor.builder()
+ .setCommand(command)
+ .setArguments(
+ Arrays.asList(portFlag, serveHttpFlag,
idleShutdownTimeoutFlag, logLevelFlag))
+ .build();
+ executor.execute();
+ if (waitForPort(executor, port)) {
+ return executor;
+ }
+ executor.stop();
+ String message =
+ String.format(
+ "Prism failed to start listening on port %d (attempt %d/%d)",
+ port, attempt, MAX_START_ATTEMPTS);
+ LOG.warn("{}", message);
+ if (attempt >= MAX_START_ATTEMPTS) {
+ throw new IOException(message);
+ }
+ }
+ }
+
+ private static boolean waitForPort(PrismExecutor executor, int port) {
+ long deadlineMillis = System.currentTimeMillis() + 10_000L;
+ while (System.currentTimeMillis() < deadlineMillis) {
+ if (!executor.isAlive()) {
+ return false;
+ }
+ try (Socket socket = new Socket("localhost", port)) {
+ return executor.isAlive();
+ } catch (IOException ignored) {
+ try {
+ Thread.sleep(10L);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ }
+ }
+ return executor.isAlive();
Review Comment:
Could this return `false` after the polling deadline? Returning
`executor.isAlive()` accepts a process that never opened its port.
--
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]
