From a022dd4ffd051fc47a00c510ee3a2eda200f1c4c Mon Sep 17 00:00:00 2001 From: Danny McCormick Date: Fri, 6 Nov 2026 14:31:17 +0000 Subject: [PATCH 1/2] Wait for expansion services to listen before running XVR tests --- .../util/construction/BaseExternalTest.java | 20 +++++-- sdks/python/scripts/run_expansion_services.sh | 56 ++++++++++++++++--- 2 files changed, 61 insertions(+), 15 deletions(-) diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/BaseExternalTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/BaseExternalTest.java index 0532f34a23b8..20c95e3ecdea 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/BaseExternalTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/BaseExternalTest.java @@ -54,14 +54,22 @@ public void tearDown() { private void waitForReady() { try { - ManagedChannel channel = ManagedChannelBuilder.forTarget(expansionAddr).build(); - ConnectivityState state = channel.getState(true); - for (int retry = 0; retry < 30 && state != ConnectivityState.READY; retry++) { - Thread.sleep(500); - state = channel.getState(true); + ManagedChannel channel = + ManagedChannelBuilder.forTarget(expansionAddr).usePlaintext().build(); + try { + ConnectivityState state = channel.getState(true); + for (int retry = 0; retry < 120 && state != ConnectivityState.READY; retry++) { + Thread.sleep(500); + if (state == ConnectivityState.TRANSIENT_FAILURE) { + channel.resetConnectBackoff(); + } + state = channel.getState(true); + } + } finally { + channel.shutdownNow(); } - channel.shutdownNow(); } catch (InterruptedException e) { + Thread.currentThread().interrupt(); throw new RuntimeException("interrupted."); } } diff --git a/sdks/python/scripts/run_expansion_services.sh b/sdks/python/scripts/run_expansion_services.sh index b59bb688ffd5..5c173d34a2c4 100755 --- a/sdks/python/scripts/run_expansion_services.sh +++ b/sdks/python/scripts/run_expansion_services.sh @@ -108,6 +108,30 @@ if [[ $CHECK_FLOCK -eq 0 ]]; then fi fi +wait_for_port() { + local port=$1 + local srv_pid=$2 + local name=$3 + local logfile=$4 + local elapsed=0 + while [ $elapsed -lt 60 ]; do + if ! kill -0 "$srv_pid" >/dev/null 2>&1; then + echo "$name expansion service (pid $srv_pid) exited prematurely. Logs:" + cat "$logfile" || true + return 1 + fi + if python -c "import socket, sys; s = socket.socket(); res = s.connect_ex(('localhost', int(sys.argv[1]))); s.close(); sys.exit(0 if res == 0 else 1)" "$port" >/dev/null 2>&1; then + echo "$name expansion service is listening @ $port" + return 0 + fi + sleep 1 + elapsed=$((elapsed + 1)) + done + echo "Timed out waiting 60s for $name expansion service @ $port. Logs:" + cat "$logfile" || true + return 1 +} + case $STARTSTOP in start) if [ -f "$pid" ]; then @@ -115,24 +139,38 @@ case $STARTSTOP in exit 0 fi + source $PYTHON_VIRTUALENV_DIR/bin/activate + echo "Launching Java expansion service @ $JAVA_PORT" - java -jar $JAVA_EXPANSION_SERVICE_JAR $JAVA_PORT --javaClassLookupAllowlistFile="$JAVA_EXPANSION_SERVICE_ALLOWLIST_FILE" >$TEMP_DIR/$FILE_BASE-java.log 2>&1 /dev/null 2>&1; then - echo $mypid >> $pid + if [[ $CHECK_FLOCK -eq 0 ]]; then + java -jar $JAVA_EXPANSION_SERVICE_JAR $JAVA_PORT --javaClassLookupAllowlistFile="$JAVA_EXPANSION_SERVICE_ALLOWLIST_FILE" >$TEMP_DIR/$FILE_BASE-java.log 2>&1 &- & + else + java -jar $JAVA_EXPANSION_SERVICE_JAR $JAVA_PORT --javaClassLookupAllowlistFile="$JAVA_EXPANSION_SERVICE_ALLOWLIST_FILE" >$TEMP_DIR/$FILE_BASE-java.log 2>&1 /dev/null 2>&1; then + echo $java_pid >> $pid else echo "Can't start Java expansion service." + exit 1 fi echo "Launching Python expansion service @ $PYTHON_PORT" - source $PYTHON_VIRTUALENV_DIR/bin/activate - python -m $PYTHON_EXPANSION_SERVICE_MODULE -p $PYTHON_PORT --fully_qualified_name_glob "$PYTHON_EXPANSION_SERVICE_ALLOWLIST_GLOB" >$TEMP_DIR/$FILE_BASE-python.log 2>&1 /dev/null 2>&1; then - echo $mypid >> $pid + if [[ $CHECK_FLOCK -eq 0 ]]; then + python -m $PYTHON_EXPANSION_SERVICE_MODULE -p $PYTHON_PORT --fully_qualified_name_glob "$PYTHON_EXPANSION_SERVICE_ALLOWLIST_GLOB" >$TEMP_DIR/$FILE_BASE-python.log 2>&1 &- & + else + python -m $PYTHON_EXPANSION_SERVICE_MODULE -p $PYTHON_PORT --fully_qualified_name_glob "$PYTHON_EXPANSION_SERVICE_ALLOWLIST_GLOB" >$TEMP_DIR/$FILE_BASE-python.log 2>&1 /dev/null 2>&1; then + echo $python_pid >> $pid else echo "Can't start Python expansion service." + exit 1 fi + + wait_for_port "$JAVA_PORT" "$java_pid" "Java" "$TEMP_DIR/$FILE_BASE-java.log" + wait_for_port "$PYTHON_PORT" "$python_pid" "Python" "$TEMP_DIR/$FILE_BASE-python.log" ;; stop) if [ -f "$pid" ]; then From c0c09bd34a6c8fe801d2481b92067670e93d88d2 Mon Sep 17 00:00:00 2001 From: Danny McCormick Date: Fri, 6 Nov 2026 17:04:16 +0000 Subject: [PATCH 2/2] Fail fast if wait_for_port returns non-zero in run_expansion_services.sh --- sdks/python/scripts/run_expansion_services.sh | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/sdks/python/scripts/run_expansion_services.sh b/sdks/python/scripts/run_expansion_services.sh index 5c173d34a2c4..9f6d1a9699e8 100755 --- a/sdks/python/scripts/run_expansion_services.sh +++ b/sdks/python/scripts/run_expansion_services.sh @@ -169,8 +169,8 @@ case $STARTSTOP in exit 1 fi - wait_for_port "$JAVA_PORT" "$java_pid" "Java" "$TEMP_DIR/$FILE_BASE-java.log" - wait_for_port "$PYTHON_PORT" "$python_pid" "Python" "$TEMP_DIR/$FILE_BASE-python.log" + wait_for_port "$JAVA_PORT" "$java_pid" "Java" "$TEMP_DIR/$FILE_BASE-java.log" || exit 1 + wait_for_port "$PYTHON_PORT" "$python_pid" "Python" "$TEMP_DIR/$FILE_BASE-python.log" || exit 1 ;; stop) if [ -f "$pid" ]; then