You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "fresh-borzoni (via GitHub)" <gi...@apache.org> on 2026/03/07 22:17:38 UTC
[PR] chore: parallelize python tests [fluss-rust]
fresh-borzoni opened a new pull request, #435:
URL: https://github.com/apache/fluss-rust/pull/435
## Summary
closes #372
Pallelize Python integration tests using pytest-xdist, matching the C++/Rust pattern. First worker starts the Docker cluster, others detect and reuse it via port check. Containers are cleaned up after all workers finish via pytest_unconfigure.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] chore: parallelize python tests [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #435:
URL: https://github.com/apache/fluss-rust/pull/435#issuecomment-4017567562
@luoyuxia @leekeiabstraction @charlesdong1991 PTAL 🙏
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] chore: parallelize python integration tests [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #435:
URL: https://github.com/apache/fluss-rust/pull/435#discussion_r2901616933
##########
bindings/python/test/conftest.py:
##########
@@ -37,47 +48,77 @@
FLUSS_VERSION = "0.9.0-incubating"
BOOTSTRAP_SERVERS_ENV = os.environ.get("FLUSS_BOOTSTRAP_SERVERS")
+# Container / network names
+NETWORK_NAME = "fluss-python-test-network"
+ZOOKEEPER_NAME = "zookeeper-python-test"
+COORDINATOR_NAME = "coordinator-server-python-test"
+TABLET_SERVER_NAME = "tablet-server-python-test"
+
+# Fixed host ports (must match across workers)
+COORDINATOR_PORT = 9123
+TABLET_SERVER_PORT = 9124
+PLAIN_CLIENT_PORT = 9223
+PLAIN_CLIENT_TABLET_PORT = 9224
+
+ALL_PORTS = [COORDINATOR_PORT, TABLET_SERVER_PORT, PLAIN_CLIENT_PORT, PLAIN_CLIENT_TABLET_PORT]
+
def _wait_for_port(host, port, timeout=60):
"""Wait for a TCP port to become available."""
start = time.time()
while time.time() - start < timeout:
try:
with socket.create_connection((host, port), timeout=1):
- return
+ return True
except (ConnectionRefusedError, TimeoutError, OSError):
time.sleep(1)
- raise TimeoutError(f"Port {port} on {host} not available after {timeout}s")
+ return False
-@pytest.fixture(scope="session")
-def fluss_cluster():
- """Start a Fluss cluster using testcontainers, or use an existing one."""
- if BOOTSTRAP_SERVERS_ENV:
- yield (BOOTSTRAP_SERVERS_ENV, BOOTSTRAP_SERVERS_ENV)
+def _all_ports_ready(timeout=60):
+ """Wait for all cluster ports to become available."""
+ deadline = time.time() + timeout
+ for port in ALL_PORTS:
+ remaining = deadline - time.time()
+ if remaining <= 0 or not _wait_for_port("localhost", port, timeout=remaining):
+ return False
+ return True
+
+
+def _run_cmd(cmd):
+ """Run a shell command, return exit code."""
+ return subprocess.run(cmd, shell=True, capture_output=True).returncode
+
+
+def _start_cluster():
+ """Start the Fluss Docker cluster via testcontainers.
+
+ If another worker already started the cluster (detected via port check),
+ reuse it. If container creation fails (name conflict from a racing worker),
+ wait for the other worker's cluster to become ready.
+ """
+ # Reuse cluster started by another parallel worker or previous run.
+ if _wait_for_port("localhost", PLAIN_CLIENT_PORT, timeout=1):
Review Comment:
not applicable
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] chore: parallelize python integration tests [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #435:
URL: https://github.com/apache/fluss-rust/pull/435#discussion_r2901729558
##########
bindings/python/test/conftest.py:
##########
@@ -37,47 +50,77 @@
FLUSS_VERSION = "0.9.0-incubating"
BOOTSTRAP_SERVERS_ENV = os.environ.get("FLUSS_BOOTSTRAP_SERVERS")
+# Container / network names
+NETWORK_NAME = "fluss-python-test-network"
+ZOOKEEPER_NAME = "zookeeper-python-test"
+COORDINATOR_NAME = "coordinator-server-python-test"
+TABLET_SERVER_NAME = "tablet-server-python-test"
+
+# Fixed host ports (must match across workers)
+COORDINATOR_PORT = 9123
+TABLET_SERVER_PORT = 9124
+PLAIN_CLIENT_PORT = 9223
+PLAIN_CLIENT_TABLET_PORT = 9224
+
+ALL_PORTS = [COORDINATOR_PORT, TABLET_SERVER_PORT, PLAIN_CLIENT_PORT, PLAIN_CLIENT_TABLET_PORT]
+
def _wait_for_port(host, port, timeout=60):
"""Wait for a TCP port to become available."""
start = time.time()
while time.time() - start < timeout:
try:
with socket.create_connection((host, port), timeout=1):
- return
+ return True
except (ConnectionRefusedError, TimeoutError, OSError):
time.sleep(1)
- raise TimeoutError(f"Port {port} on {host} not available after {timeout}s")
+ return False
-@pytest.fixture(scope="session")
-def fluss_cluster():
- """Start a Fluss cluster using testcontainers, or use an existing one."""
- if BOOTSTRAP_SERVERS_ENV:
- yield (BOOTSTRAP_SERVERS_ENV, BOOTSTRAP_SERVERS_ENV)
+def _all_ports_ready(timeout=60):
+ """Wait for all cluster ports to become available."""
+ deadline = time.time() + timeout
+ for port in ALL_PORTS:
+ remaining = deadline - time.time()
+ if remaining <= 0 or not _wait_for_port("localhost", port, timeout=remaining):
+ return False
+ return True
+
+
+def _run_cmd(cmd):
+ """Run a command (list form), return exit code."""
+ return subprocess.run(cmd, capture_output=True).returncode
+
+
+def _start_cluster():
+ """Start the Fluss Docker cluster via testcontainers.
+
+ If another worker already started the cluster (detected via port check),
+ reuse it. If container creation fails (name conflict from a racing worker),
+ wait for the other worker's cluster to become ready.
+ """
+ # Reuse cluster started by another parallel worker or previous run.
+ if _wait_for_port("localhost", PLAIN_CLIENT_PORT, timeout=1):
Review Comment:
it's a fast check, so we are just quickly checking if there is warm cluster running already, usual integration CI/CD would always have this fast path failed.
For local dev if i wish to quickly iterate on tests, i just comment out cleanup and the next run would reuse existing cluster and run superfast, as the cluster is already warm.
One improvement: we may use some env var and wire it in a such way that we can specify to retain cluster and skip cleanup. But it's followup, the main focus was integration tests.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] chore: parallelize python integration tests [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #435:
URL: https://github.com/apache/fluss-rust/pull/435#discussion_r2901729558
##########
bindings/python/test/conftest.py:
##########
@@ -37,47 +50,77 @@
FLUSS_VERSION = "0.9.0-incubating"
BOOTSTRAP_SERVERS_ENV = os.environ.get("FLUSS_BOOTSTRAP_SERVERS")
+# Container / network names
+NETWORK_NAME = "fluss-python-test-network"
+ZOOKEEPER_NAME = "zookeeper-python-test"
+COORDINATOR_NAME = "coordinator-server-python-test"
+TABLET_SERVER_NAME = "tablet-server-python-test"
+
+# Fixed host ports (must match across workers)
+COORDINATOR_PORT = 9123
+TABLET_SERVER_PORT = 9124
+PLAIN_CLIENT_PORT = 9223
+PLAIN_CLIENT_TABLET_PORT = 9224
+
+ALL_PORTS = [COORDINATOR_PORT, TABLET_SERVER_PORT, PLAIN_CLIENT_PORT, PLAIN_CLIENT_TABLET_PORT]
+
def _wait_for_port(host, port, timeout=60):
"""Wait for a TCP port to become available."""
start = time.time()
while time.time() - start < timeout:
try:
with socket.create_connection((host, port), timeout=1):
- return
+ return True
except (ConnectionRefusedError, TimeoutError, OSError):
time.sleep(1)
- raise TimeoutError(f"Port {port} on {host} not available after {timeout}s")
+ return False
-@pytest.fixture(scope="session")
-def fluss_cluster():
- """Start a Fluss cluster using testcontainers, or use an existing one."""
- if BOOTSTRAP_SERVERS_ENV:
- yield (BOOTSTRAP_SERVERS_ENV, BOOTSTRAP_SERVERS_ENV)
+def _all_ports_ready(timeout=60):
+ """Wait for all cluster ports to become available."""
+ deadline = time.time() + timeout
+ for port in ALL_PORTS:
+ remaining = deadline - time.time()
+ if remaining <= 0 or not _wait_for_port("localhost", port, timeout=remaining):
+ return False
+ return True
+
+
+def _run_cmd(cmd):
+ """Run a command (list form), return exit code."""
+ return subprocess.run(cmd, capture_output=True).returncode
+
+
+def _start_cluster():
+ """Start the Fluss Docker cluster via testcontainers.
+
+ If another worker already started the cluster (detected via port check),
+ reuse it. If container creation fails (name conflict from a racing worker),
+ wait for the other worker's cluster to become ready.
+ """
+ # Reuse cluster started by another parallel worker or previous run.
+ if _wait_for_port("localhost", PLAIN_CLIENT_PORT, timeout=1):
Review Comment:
it's a fast check, so we are just quickly checking if there is warm cluster running already, usual integration CI/CD would always have this fast path failed.
For local dev if i wish to quickly iterate on tests, i just comment out cleanup and the next run would reuse existing cluster and run superfast, as the cluster is already warm. So I left this hook if someone wishes to use it as well.
One improvement: we may use some env var and wire it in a such way that we can specify to retain cluster and skip cleanup. But it's followup, the main focus was integration tests.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] chore: parallelize python integration tests [fluss-rust]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #435:
URL: https://github.com/apache/fluss-rust/pull/435#discussion_r2901160043
##########
bindings/python/test/conftest.py:
##########
@@ -20,14 +20,25 @@
If FLUSS_BOOTSTRAP_SERVERS is set, tests connect to an existing cluster.
Otherwise, a Fluss cluster is started automatically via testcontainers.
+The first pytest-xdist worker to run starts the cluster; other workers
+detect it via port check and reuse it (matching the C++ test pattern).
+Containers are cleaned up after all workers finish via pytest_unconfigure.
+
Run with:
- uv run maturin develop && uv run pytest test/ -v
+ uv run maturin develop && uv run pytest test/ -v -n auto
"""
+import asyncio
import os
import socket
+import subprocess
import time
+# Disable testcontainers Ryuk reaper — it would kill containers when the
+# first xdist worker exits, while other workers are still running.
+# We handle cleanup ourselves in pytest_unconfigure.
+os.environ.setdefault("TESTCONTAINERS_RYUK_DISABLED", "true")
Review Comment:
Ryuk is disabled unconditionally at import time. That changes behavior for non-xdist runs too and can leave orphaned containers if a local single-process test run is interrupted before pytest_unconfigure executes. Consider disabling Ryuk only when xdist is active (e.g., when PYTEST_XDIST_WORKER is set), and otherwise keep the default reaper behavior.
```suggestion
# Disable testcontainers Ryuk reaper for xdist workers — it would kill
# containers when the first xdist worker exits, while other workers are
# still running. We handle cleanup ourselves in pytest_unconfigure.
if os.environ.get("PYTEST_XDIST_WORKER"):
os.environ.setdefault("TESTCONTAINERS_RYUK_DISABLED", "true")
```
##########
bindings/python/test/conftest.py:
##########
@@ -37,47 +48,77 @@
FLUSS_VERSION = "0.9.0-incubating"
BOOTSTRAP_SERVERS_ENV = os.environ.get("FLUSS_BOOTSTRAP_SERVERS")
+# Container / network names
+NETWORK_NAME = "fluss-python-test-network"
+ZOOKEEPER_NAME = "zookeeper-python-test"
+COORDINATOR_NAME = "coordinator-server-python-test"
+TABLET_SERVER_NAME = "tablet-server-python-test"
+
+# Fixed host ports (must match across workers)
+COORDINATOR_PORT = 9123
+TABLET_SERVER_PORT = 9124
+PLAIN_CLIENT_PORT = 9223
+PLAIN_CLIENT_TABLET_PORT = 9224
+
+ALL_PORTS = [COORDINATOR_PORT, TABLET_SERVER_PORT, PLAIN_CLIENT_PORT, PLAIN_CLIENT_TABLET_PORT]
+
def _wait_for_port(host, port, timeout=60):
"""Wait for a TCP port to become available."""
start = time.time()
while time.time() - start < timeout:
try:
with socket.create_connection((host, port), timeout=1):
- return
+ return True
except (ConnectionRefusedError, TimeoutError, OSError):
time.sleep(1)
- raise TimeoutError(f"Port {port} on {host} not available after {timeout}s")
+ return False
-@pytest.fixture(scope="session")
-def fluss_cluster():
- """Start a Fluss cluster using testcontainers, or use an existing one."""
- if BOOTSTRAP_SERVERS_ENV:
- yield (BOOTSTRAP_SERVERS_ENV, BOOTSTRAP_SERVERS_ENV)
+def _all_ports_ready(timeout=60):
+ """Wait for all cluster ports to become available."""
+ deadline = time.time() + timeout
+ for port in ALL_PORTS:
+ remaining = deadline - time.time()
+ if remaining <= 0 or not _wait_for_port("localhost", port, timeout=remaining):
+ return False
+ return True
+
+
+def _run_cmd(cmd):
+ """Run a shell command, return exit code."""
+ return subprocess.run(cmd, shell=True, capture_output=True).returncode
+
+
+def _start_cluster():
+ """Start the Fluss Docker cluster via testcontainers.
+
+ If another worker already started the cluster (detected via port check),
+ reuse it. If container creation fails (name conflict from a racing worker),
+ wait for the other worker's cluster to become ready.
+ """
+ # Reuse cluster started by another parallel worker or previous run.
+ if _wait_for_port("localhost", PLAIN_CLIENT_PORT, timeout=1):
Review Comment:
The reuse path returns as soon as the plaintext port is connectable, but other listeners (and/or TabletServer readiness) may still be initializing. Since some tests create their own connections without retry logic, this can make parallel runs flaky. Consider reusing only after _all_ports_ready() (or a lightweight Fluss-level readiness check) succeeds instead of checking a single port.
```suggestion
# Only reuse once all expected ports are ready, to avoid flakiness when
# other listeners or TabletServer are still initializing.
if _all_ports_ready(timeout=5):
```
##########
bindings/python/test/conftest.py:
##########
@@ -37,47 +48,77 @@
FLUSS_VERSION = "0.9.0-incubating"
BOOTSTRAP_SERVERS_ENV = os.environ.get("FLUSS_BOOTSTRAP_SERVERS")
+# Container / network names
+NETWORK_NAME = "fluss-python-test-network"
+ZOOKEEPER_NAME = "zookeeper-python-test"
+COORDINATOR_NAME = "coordinator-server-python-test"
+TABLET_SERVER_NAME = "tablet-server-python-test"
+
+# Fixed host ports (must match across workers)
+COORDINATOR_PORT = 9123
+TABLET_SERVER_PORT = 9124
+PLAIN_CLIENT_PORT = 9223
+PLAIN_CLIENT_TABLET_PORT = 9224
+
+ALL_PORTS = [COORDINATOR_PORT, TABLET_SERVER_PORT, PLAIN_CLIENT_PORT, PLAIN_CLIENT_TABLET_PORT]
+
def _wait_for_port(host, port, timeout=60):
"""Wait for a TCP port to become available."""
start = time.time()
while time.time() - start < timeout:
try:
with socket.create_connection((host, port), timeout=1):
- return
+ return True
except (ConnectionRefusedError, TimeoutError, OSError):
time.sleep(1)
- raise TimeoutError(f"Port {port} on {host} not available after {timeout}s")
+ return False
-@pytest.fixture(scope="session")
-def fluss_cluster():
- """Start a Fluss cluster using testcontainers, or use an existing one."""
- if BOOTSTRAP_SERVERS_ENV:
- yield (BOOTSTRAP_SERVERS_ENV, BOOTSTRAP_SERVERS_ENV)
+def _all_ports_ready(timeout=60):
+ """Wait for all cluster ports to become available."""
+ deadline = time.time() + timeout
+ for port in ALL_PORTS:
+ remaining = deadline - time.time()
+ if remaining <= 0 or not _wait_for_port("localhost", port, timeout=remaining):
+ return False
+ return True
+
+
+def _run_cmd(cmd):
+ """Run a shell command, return exit code."""
+ return subprocess.run(cmd, shell=True, capture_output=True).returncode
+
+
+def _start_cluster():
+ """Start the Fluss Docker cluster via testcontainers.
+
+ If another worker already started the cluster (detected via port check),
+ reuse it. If container creation fails (name conflict from a racing worker),
+ wait for the other worker's cluster to become ready.
+ """
+ # Reuse cluster started by another parallel worker or previous run.
+ if _wait_for_port("localhost", PLAIN_CLIENT_PORT, timeout=1):
+ print("Reusing existing cluster via port check.")
return
from testcontainers.core.container import DockerContainer
- from testcontainers.core.network import Network
- network = Network()
- network.create()
+ print("Starting Fluss cluster via testcontainers...")
- zookeeper = (
- DockerContainer("zookeeper:3.9.2")
- .with_network(network)
- .with_name("zookeeper-python-test")
- )
+ # Create a named network via Docker CLI (idempotent, avoids orphaned
+ # random-named networks when multiple xdist workers race).
+ _run_cmd(f"docker network create {NETWORK_NAME} 2>/dev/null || true")
Review Comment:
The `docker network create ... 2>/dev/null || true` invocation is POSIX-shell specific and depends on shell redirection/`||`. If these tests are expected to run on non-POSIX shells (e.g., Windows), this will fail. Consider performing the idempotent network creation using `subprocess.run([...])` and handling the 'already exists' error code/message explicitly, without shell operators.
##########
bindings/python/test/conftest.py:
##########
@@ -37,47 +48,77 @@
FLUSS_VERSION = "0.9.0-incubating"
BOOTSTRAP_SERVERS_ENV = os.environ.get("FLUSS_BOOTSTRAP_SERVERS")
+# Container / network names
+NETWORK_NAME = "fluss-python-test-network"
+ZOOKEEPER_NAME = "zookeeper-python-test"
+COORDINATOR_NAME = "coordinator-server-python-test"
+TABLET_SERVER_NAME = "tablet-server-python-test"
+
+# Fixed host ports (must match across workers)
+COORDINATOR_PORT = 9123
+TABLET_SERVER_PORT = 9124
+PLAIN_CLIENT_PORT = 9223
+PLAIN_CLIENT_TABLET_PORT = 9224
+
+ALL_PORTS = [COORDINATOR_PORT, TABLET_SERVER_PORT, PLAIN_CLIENT_PORT, PLAIN_CLIENT_TABLET_PORT]
+
def _wait_for_port(host, port, timeout=60):
"""Wait for a TCP port to become available."""
start = time.time()
while time.time() - start < timeout:
try:
with socket.create_connection((host, port), timeout=1):
- return
+ return True
except (ConnectionRefusedError, TimeoutError, OSError):
time.sleep(1)
- raise TimeoutError(f"Port {port} on {host} not available after {timeout}s")
+ return False
-@pytest.fixture(scope="session")
-def fluss_cluster():
- """Start a Fluss cluster using testcontainers, or use an existing one."""
- if BOOTSTRAP_SERVERS_ENV:
- yield (BOOTSTRAP_SERVERS_ENV, BOOTSTRAP_SERVERS_ENV)
+def _all_ports_ready(timeout=60):
+ """Wait for all cluster ports to become available."""
+ deadline = time.time() + timeout
+ for port in ALL_PORTS:
+ remaining = deadline - time.time()
+ if remaining <= 0 or not _wait_for_port("localhost", port, timeout=remaining):
+ return False
+ return True
+
+
+def _run_cmd(cmd):
+ """Run a shell command, return exit code."""
+ return subprocess.run(cmd, shell=True, capture_output=True).returncode
+
+
Review Comment:
_run_cmd uses shell=True and discards stdout/stderr. Even though the current command is built from constants, shell=True plus suppressed output makes failures harder to diagnose and relies on shell semantics. Prefer running docker via an argv list (no shell) and include stderr in the error path (or raise when the command fails).
```suggestion
"""Run a command, return exit code.
The command is executed without a shell. On failure, stderr is printed
to aid debugging while preserving the return-code-based API.
"""
# Support both string and list commands without relying on shell parsing.
if isinstance(cmd, str):
# Simple whitespace split is sufficient for constant commands used here.
cmd_to_run = cmd.split()
else:
cmd_to_run = cmd
result = subprocess.run(cmd_to_run, shell=False, capture_output=True, text=True)
if result.returncode != 0:
print(f"Command {cmd!r} failed with exit code {result.returncode}")
if result.stderr:
print(result.stderr)
return result.returncode
```
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] chore: parallelize python integration tests [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #435:
URL: https://github.com/apache/fluss-rust/pull/435#discussion_r2901617647
##########
bindings/python/test/conftest.py:
##########
@@ -20,14 +20,25 @@
If FLUSS_BOOTSTRAP_SERVERS is set, tests connect to an existing cluster.
Otherwise, a Fluss cluster is started automatically via testcontainers.
+The first pytest-xdist worker to run starts the cluster; other workers
+detect it via port check and reuse it (matching the C++ test pattern).
+Containers are cleaned up after all workers finish via pytest_unconfigure.
+
Run with:
- uv run maturin develop && uv run pytest test/ -v
+ uv run maturin develop && uv run pytest test/ -v -n auto
"""
+import asyncio
import os
import socket
+import subprocess
import time
+# Disable testcontainers Ryuk reaper — it would kill containers when the
+# first xdist worker exits, while other workers are still running.
+# We handle cleanup ourselves in pytest_unconfigure.
+os.environ.setdefault("TESTCONTAINERS_RYUK_DISABLED", "true")
Review Comment:
minor, but yes, we can disable Ryuk when no XDIST
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] chore: parallelize python integration tests [fluss-rust]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia merged PR #435:
URL: https://github.com/apache/fluss-rust/pull/435
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] chore: parallelize python integration tests [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #435:
URL: https://github.com/apache/fluss-rust/pull/435#issuecomment-4018756692
Addressed comments
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] chore: parallelize python integration tests [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #435:
URL: https://github.com/apache/fluss-rust/pull/435#discussion_r2901734880
##########
bindings/python/test/conftest.py:
##########
@@ -37,47 +50,77 @@
FLUSS_VERSION = "0.9.0-incubating"
BOOTSTRAP_SERVERS_ENV = os.environ.get("FLUSS_BOOTSTRAP_SERVERS")
+# Container / network names
+NETWORK_NAME = "fluss-python-test-network"
+ZOOKEEPER_NAME = "zookeeper-python-test"
+COORDINATOR_NAME = "coordinator-server-python-test"
+TABLET_SERVER_NAME = "tablet-server-python-test"
+
+# Fixed host ports (must match across workers)
+COORDINATOR_PORT = 9123
+TABLET_SERVER_PORT = 9124
+PLAIN_CLIENT_PORT = 9223
+PLAIN_CLIENT_TABLET_PORT = 9224
+
+ALL_PORTS = [COORDINATOR_PORT, TABLET_SERVER_PORT, PLAIN_CLIENT_PORT, PLAIN_CLIENT_TABLET_PORT]
+
def _wait_for_port(host, port, timeout=60):
"""Wait for a TCP port to become available."""
start = time.time()
while time.time() - start < timeout:
try:
with socket.create_connection((host, port), timeout=1):
- return
+ return True
except (ConnectionRefusedError, TimeoutError, OSError):
time.sleep(1)
- raise TimeoutError(f"Port {port} on {host} not available after {timeout}s")
+ return False
-@pytest.fixture(scope="session")
-def fluss_cluster():
- """Start a Fluss cluster using testcontainers, or use an existing one."""
- if BOOTSTRAP_SERVERS_ENV:
- yield (BOOTSTRAP_SERVERS_ENV, BOOTSTRAP_SERVERS_ENV)
+def _all_ports_ready(timeout=60):
+ """Wait for all cluster ports to become available."""
+ deadline = time.time() + timeout
+ for port in ALL_PORTS:
+ remaining = deadline - time.time()
+ if remaining <= 0 or not _wait_for_port("localhost", port, timeout=remaining):
+ return False
+ return True
+
+
+def _run_cmd(cmd):
+ """Run a command (list form), return exit code."""
+ return subprocess.run(cmd, capture_output=True).returncode
Review Comment:
for now it's used for docker network call only, but valid observation 👍
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] chore: parallelize python integration tests [fluss-rust]
Posted by "charlesdong1991 (via GitHub)" <gi...@apache.org>.
charlesdong1991 commented on code in PR #435:
URL: https://github.com/apache/fluss-rust/pull/435#discussion_r2901654842
##########
bindings/python/test/conftest.py:
##########
@@ -112,42 +145,121 @@ def fluss_cluster():
"netty.server.num-network-threads: 1",
"netty.server.num-worker-threads: 3",
])
+
+ zookeeper = (
+ DockerContainer("zookeeper:3.9.2")
+ .with_kwargs(network=NETWORK_NAME)
+ .with_name(ZOOKEEPER_NAME)
+ )
+ coordinator = (
+ DockerContainer(f"{FLUSS_IMAGE}:{FLUSS_VERSION}")
+ .with_kwargs(network=NETWORK_NAME)
+ .with_name(COORDINATOR_NAME)
+ .with_bind_ports(9123, 9123)
+ .with_bind_ports(9223, 9223)
+ .with_command("coordinatorServer")
+ .with_env("FLUSS_PROPERTIES", coordinator_props)
+ )
tablet_server = (
DockerContainer(f"{FLUSS_IMAGE}:{FLUSS_VERSION}")
- .with_network(network)
- .with_name("tablet-server-python-test")
+ .with_kwargs(network=NETWORK_NAME)
+ .with_name(TABLET_SERVER_NAME)
.with_bind_ports(9123, 9124)
.with_bind_ports(9223, 9224)
.with_command("tabletServer")
.with_env("FLUSS_PROPERTIES", tablet_props)
)
- zookeeper.start()
- coordinator.start()
- tablet_server.start()
+ try:
+ zookeeper.start()
+ coordinator.start()
+ tablet_server.start()
+ except Exception as e:
Review Comment:
this is a bit broad here, maybe consider narrowing to docker specific errors?
btw IIUC, we want to start fluss clsuter exactly once per test session
according to pytest-xdist: https://pytest-xdist.readthedocs.io/en/latest/how-to.html#making-session-scoped-fixtures-execute-only-once , they recommend to use file lock to coordinate startup
##########
bindings/python/test/conftest.py:
##########
@@ -37,47 +50,77 @@
FLUSS_VERSION = "0.9.0-incubating"
BOOTSTRAP_SERVERS_ENV = os.environ.get("FLUSS_BOOTSTRAP_SERVERS")
+# Container / network names
+NETWORK_NAME = "fluss-python-test-network"
+ZOOKEEPER_NAME = "zookeeper-python-test"
+COORDINATOR_NAME = "coordinator-server-python-test"
+TABLET_SERVER_NAME = "tablet-server-python-test"
+
+# Fixed host ports (must match across workers)
+COORDINATOR_PORT = 9123
+TABLET_SERVER_PORT = 9124
+PLAIN_CLIENT_PORT = 9223
+PLAIN_CLIENT_TABLET_PORT = 9224
+
+ALL_PORTS = [COORDINATOR_PORT, TABLET_SERVER_PORT, PLAIN_CLIENT_PORT, PLAIN_CLIENT_TABLET_PORT]
+
def _wait_for_port(host, port, timeout=60):
"""Wait for a TCP port to become available."""
start = time.time()
while time.time() - start < timeout:
try:
with socket.create_connection((host, port), timeout=1):
- return
+ return True
except (ConnectionRefusedError, TimeoutError, OSError):
time.sleep(1)
- raise TimeoutError(f"Port {port} on {host} not available after {timeout}s")
+ return False
-@pytest.fixture(scope="session")
-def fluss_cluster():
- """Start a Fluss cluster using testcontainers, or use an existing one."""
- if BOOTSTRAP_SERVERS_ENV:
- yield (BOOTSTRAP_SERVERS_ENV, BOOTSTRAP_SERVERS_ENV)
+def _all_ports_ready(timeout=60):
+ """Wait for all cluster ports to become available."""
+ deadline = time.time() + timeout
+ for port in ALL_PORTS:
+ remaining = deadline - time.time()
+ if remaining <= 0 or not _wait_for_port("localhost", port, timeout=remaining):
+ return False
+ return True
+
+
+def _run_cmd(cmd):
+ """Run a command (list form), return exit code."""
+ return subprocess.run(cmd, capture_output=True).returncode
Review Comment:
if run fails, we will only see error code, but not the error message now, not sure if it will hinder debugging if needed?
##########
bindings/python/test/conftest.py:
##########
@@ -37,47 +50,77 @@
FLUSS_VERSION = "0.9.0-incubating"
BOOTSTRAP_SERVERS_ENV = os.environ.get("FLUSS_BOOTSTRAP_SERVERS")
+# Container / network names
+NETWORK_NAME = "fluss-python-test-network"
+ZOOKEEPER_NAME = "zookeeper-python-test"
+COORDINATOR_NAME = "coordinator-server-python-test"
+TABLET_SERVER_NAME = "tablet-server-python-test"
+
+# Fixed host ports (must match across workers)
+COORDINATOR_PORT = 9123
+TABLET_SERVER_PORT = 9124
+PLAIN_CLIENT_PORT = 9223
+PLAIN_CLIENT_TABLET_PORT = 9224
+
+ALL_PORTS = [COORDINATOR_PORT, TABLET_SERVER_PORT, PLAIN_CLIENT_PORT, PLAIN_CLIENT_TABLET_PORT]
+
def _wait_for_port(host, port, timeout=60):
"""Wait for a TCP port to become available."""
start = time.time()
while time.time() - start < timeout:
try:
with socket.create_connection((host, port), timeout=1):
- return
+ return True
except (ConnectionRefusedError, TimeoutError, OSError):
time.sleep(1)
- raise TimeoutError(f"Port {port} on {host} not available after {timeout}s")
+ return False
-@pytest.fixture(scope="session")
-def fluss_cluster():
- """Start a Fluss cluster using testcontainers, or use an existing one."""
- if BOOTSTRAP_SERVERS_ENV:
- yield (BOOTSTRAP_SERVERS_ENV, BOOTSTRAP_SERVERS_ENV)
+def _all_ports_ready(timeout=60):
+ """Wait for all cluster ports to become available."""
+ deadline = time.time() + timeout
+ for port in ALL_PORTS:
+ remaining = deadline - time.time()
+ if remaining <= 0 or not _wait_for_port("localhost", port, timeout=remaining):
+ return False
+ return True
+
+
+def _run_cmd(cmd):
+ """Run a command (list form), return exit code."""
+ return subprocess.run(cmd, capture_output=True).returncode
+
+
+def _start_cluster():
+ """Start the Fluss Docker cluster via testcontainers.
+
+ If another worker already started the cluster (detected via port check),
+ reuse it. If container creation fails (name conflict from a racing worker),
+ wait for the other worker's cluster to become ready.
+ """
+ # Reuse cluster started by another parallel worker or previous run.
+ if _wait_for_port("localhost", PLAIN_CLIENT_PORT, timeout=1):
Review Comment:
i am curious if 1s too short?
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] chore: parallelize python integration tests [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #435:
URL: https://github.com/apache/fluss-rust/pull/435#discussion_r2901713733
##########
bindings/python/test/conftest.py:
##########
@@ -112,42 +145,121 @@ def fluss_cluster():
"netty.server.num-network-threads: 1",
"netty.server.num-worker-threads: 3",
])
+
+ zookeeper = (
+ DockerContainer("zookeeper:3.9.2")
+ .with_kwargs(network=NETWORK_NAME)
+ .with_name(ZOOKEEPER_NAME)
+ )
+ coordinator = (
+ DockerContainer(f"{FLUSS_IMAGE}:{FLUSS_VERSION}")
+ .with_kwargs(network=NETWORK_NAME)
+ .with_name(COORDINATOR_NAME)
+ .with_bind_ports(9123, 9123)
+ .with_bind_ports(9223, 9223)
+ .with_command("coordinatorServer")
+ .with_env("FLUSS_PROPERTIES", coordinator_props)
+ )
tablet_server = (
DockerContainer(f"{FLUSS_IMAGE}:{FLUSS_VERSION}")
- .with_network(network)
- .with_name("tablet-server-python-test")
+ .with_kwargs(network=NETWORK_NAME)
+ .with_name(TABLET_SERVER_NAME)
.with_bind_ports(9123, 9124)
.with_bind_ports(9223, 9224)
.with_command("tabletServer")
.with_env("FLUSS_PROPERTIES", tablet_props)
)
- zookeeper.start()
- coordinator.start()
- tablet_server.start()
+ try:
+ zookeeper.start()
+ coordinator.start()
+ tablet_server.start()
+ except Exception as e:
Review Comment:
Good call 👍
file lock approcach is cleaner, but it requires more restructuring, so I just used the same approach as in cpp for consistency.
Let's create follow up if someone wishes to improve this and play with XDIST.
I don't expect tests time to change though :)
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] chore: parallelize python integration tests [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #435:
URL: https://github.com/apache/fluss-rust/pull/435#discussion_r2901729558
##########
bindings/python/test/conftest.py:
##########
@@ -37,47 +50,77 @@
FLUSS_VERSION = "0.9.0-incubating"
BOOTSTRAP_SERVERS_ENV = os.environ.get("FLUSS_BOOTSTRAP_SERVERS")
+# Container / network names
+NETWORK_NAME = "fluss-python-test-network"
+ZOOKEEPER_NAME = "zookeeper-python-test"
+COORDINATOR_NAME = "coordinator-server-python-test"
+TABLET_SERVER_NAME = "tablet-server-python-test"
+
+# Fixed host ports (must match across workers)
+COORDINATOR_PORT = 9123
+TABLET_SERVER_PORT = 9124
+PLAIN_CLIENT_PORT = 9223
+PLAIN_CLIENT_TABLET_PORT = 9224
+
+ALL_PORTS = [COORDINATOR_PORT, TABLET_SERVER_PORT, PLAIN_CLIENT_PORT, PLAIN_CLIENT_TABLET_PORT]
+
def _wait_for_port(host, port, timeout=60):
"""Wait for a TCP port to become available."""
start = time.time()
while time.time() - start < timeout:
try:
with socket.create_connection((host, port), timeout=1):
- return
+ return True
except (ConnectionRefusedError, TimeoutError, OSError):
time.sleep(1)
- raise TimeoutError(f"Port {port} on {host} not available after {timeout}s")
+ return False
-@pytest.fixture(scope="session")
-def fluss_cluster():
- """Start a Fluss cluster using testcontainers, or use an existing one."""
- if BOOTSTRAP_SERVERS_ENV:
- yield (BOOTSTRAP_SERVERS_ENV, BOOTSTRAP_SERVERS_ENV)
+def _all_ports_ready(timeout=60):
+ """Wait for all cluster ports to become available."""
+ deadline = time.time() + timeout
+ for port in ALL_PORTS:
+ remaining = deadline - time.time()
+ if remaining <= 0 or not _wait_for_port("localhost", port, timeout=remaining):
+ return False
+ return True
+
+
+def _run_cmd(cmd):
+ """Run a command (list form), return exit code."""
+ return subprocess.run(cmd, capture_output=True).returncode
+
+
+def _start_cluster():
+ """Start the Fluss Docker cluster via testcontainers.
+
+ If another worker already started the cluster (detected via port check),
+ reuse it. If container creation fails (name conflict from a racing worker),
+ wait for the other worker's cluster to become ready.
+ """
+ # Reuse cluster started by another parallel worker or previous run.
+ if _wait_for_port("localhost", PLAIN_CLIENT_PORT, timeout=1):
Review Comment:
it's a fast check, so we are just quickly checking if there is warm cluster running already, usual integration CI/CD would always have this fast path failed.
For local dev if i wish to quickly iterate on tests, i just comment out cleanup and the next run would reuse existing cluster and run superfast, as the cluster is already warm. So I left this hooks if someone wishes to use it as well.
One improvement: we may use some env var and wire it in a such way that we can specify to retain cluster and skip cleanup. But it's followup, the main focus was integration tests.
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] chore: parallelize python integration tests [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #435:
URL: https://github.com/apache/fluss-rust/pull/435#discussion_r2901734880
##########
bindings/python/test/conftest.py:
##########
@@ -37,47 +50,77 @@
FLUSS_VERSION = "0.9.0-incubating"
BOOTSTRAP_SERVERS_ENV = os.environ.get("FLUSS_BOOTSTRAP_SERVERS")
+# Container / network names
+NETWORK_NAME = "fluss-python-test-network"
+ZOOKEEPER_NAME = "zookeeper-python-test"
+COORDINATOR_NAME = "coordinator-server-python-test"
+TABLET_SERVER_NAME = "tablet-server-python-test"
+
+# Fixed host ports (must match across workers)
+COORDINATOR_PORT = 9123
+TABLET_SERVER_PORT = 9124
+PLAIN_CLIENT_PORT = 9223
+PLAIN_CLIENT_TABLET_PORT = 9224
+
+ALL_PORTS = [COORDINATOR_PORT, TABLET_SERVER_PORT, PLAIN_CLIENT_PORT, PLAIN_CLIENT_TABLET_PORT]
+
def _wait_for_port(host, port, timeout=60):
"""Wait for a TCP port to become available."""
start = time.time()
while time.time() - start < timeout:
try:
with socket.create_connection((host, port), timeout=1):
- return
+ return True
except (ConnectionRefusedError, TimeoutError, OSError):
time.sleep(1)
- raise TimeoutError(f"Port {port} on {host} not available after {timeout}s")
+ return False
-@pytest.fixture(scope="session")
-def fluss_cluster():
- """Start a Fluss cluster using testcontainers, or use an existing one."""
- if BOOTSTRAP_SERVERS_ENV:
- yield (BOOTSTRAP_SERVERS_ENV, BOOTSTRAP_SERVERS_ENV)
+def _all_ports_ready(timeout=60):
+ """Wait for all cluster ports to become available."""
+ deadline = time.time() + timeout
+ for port in ALL_PORTS:
+ remaining = deadline - time.time()
+ if remaining <= 0 or not _wait_for_port("localhost", port, timeout=remaining):
+ return False
+ return True
+
+
+def _run_cmd(cmd):
+ """Run a command (list form), return exit code."""
+ return subprocess.run(cmd, capture_output=True).returncode
Review Comment:
for now it's used for docker network call only, so not much to debug, but valid observation if used for more stuff later 👍
--
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: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org