feat: enhance ScyllaDB adapter with Docker support and update dependencies

- Add address translator for Docker/NAT environments

- Configure connection pool limits per host

- Update dependency versions to match uv.lock

- Update test containers and config template

Author: HosseinNejatiJavaremi

.env.test

@@ -32,7 +32,7 @@ REDIS__DATABASE=0
 REDIS__PASSWORD=test_password
 
 # PostgreSQL Configuration
-POSTGRES__IMAGE=postgres:18.1-alpine
+POSTGRES__IMAGE=postgres:18.2-alpine
 POSTGRES_SQLALCHEMY__HOST=localhost
 POSTGRES_SQLALCHEMY__PORT=5432
 POSTGRES_SQLALCHEMY__DATABASE=test_db
@@ -67,7 +67,7 @@ ELASTIC__HTTP_USER_NAME=elastic
 ELASTIC__HTTP_PASSWORD=test_password
 
 # Kafka Configuration
-KAFKA__IMAGE=confluentinc/cp-kafka:7.9.3
+KAFKA__IMAGE=confluentinc/cp-kafka:8.0.3
 KAFKA__BROKERS_LIST=["localhost:9092"]
 
 # MinIO Configuration
@@ -77,9 +77,10 @@ MINIO__ACCESS_KEY=test_access_key
 MINIO__SECRET_KEY=test_secret_key
 
 # ScyllaDB Configuration
-SCYLLADB__IMAGE=scylladb/scylla:2025.3
+SCYLLADB__IMAGE=scylladb/scylla:2025.4
 SCYLLADB__CONTACT_POINTS=["localhost"]
 SCYLLADB__PORT=9042
 SCYLLADB__PROTOCOL_VERSION=4
 SCYLLADB__COMPRESSION=true
 SCYLLADB__DISABLE_SHARD_AWARENESS=true
+SCYLLADB__ADDRESS_TRANSLATION_ENABLED=true

archipy/adapters/scylladb/adapters.py

@@ -14,6 +14,8 @@
 from cassandra.auth import PlainTextAuthProvider
 from cassandra.cluster import Cluster
 from cassandra.policies import (
+    AddressTranslator,
+    DCAwareRoundRobinPolicy,
     ExponentialBackoffRetryPolicy,
     FallthroughRetryPolicy,
     RoundRobinPolicy,
@@ -40,6 +42,19 @@
 logger = logging.getLogger(__name__)
 
 
+class _FixedAddressTranslator(AddressTranslator):
+    """Translates all discovered node addresses to a fixed host address.
+
+    Used in Docker/NAT environments where gossip-discovered internal IPs are unreachable.
+    """
+
+    def __init__(self, address: str) -> None:
+        self._address = address
+
+    def translate(self, addr: str) -> str:  # noqa: ARG002
+        return self._address
+
+
 class ScyllaDBExceptionHandlerMixin:
     """Mixin class to handle ScyllaDB/Cassandra exceptions in a consistent way."""
 
@@ -182,13 +197,10 @@ def _create_cluster(self) -> Any:
 
         # Configure load balancing policy with optional datacenter awareness
         if self.config.LOCAL_DC:
-            from cassandra.policies import DCAwareRoundRobinPolicy
-
             base_policy = DCAwareRoundRobinPolicy(local_dc=self.config.LOCAL_DC)
+            load_balancing_policy = TokenAwarePolicy(base_policy)
         else:
-            base_policy = RoundRobinPolicy()
-
-        load_balancing_policy = TokenAwarePolicy(base_policy)
+            load_balancing_policy = TokenAwarePolicy(RoundRobinPolicy())
 
         if self.config.RETRY_POLICY == "FALLTHROUGH":
             retry_policy = FallthroughRetryPolicy()
@@ -203,6 +215,12 @@ def _create_cluster(self) -> Any:
         if self.config.DISABLE_SHARD_AWARENESS:
             shard_aware_options = {"disable": True}
 
+        # Address translation for Docker/NAT environments where gossip-discovered
+        # internal container IPs are unreachable from the host
+        address_translator = None
+        if self.config.ADDRESS_TRANSLATION_ENABLED:
+            address_translator = _FixedAddressTranslator(self.config.CONTACT_POINTS[0])
+
         # Cluster is from cassandra.cluster, properly typed
         cluster = Cluster(
             contact_points=self.config.CONTACT_POINTS,
@@ -214,12 +232,17 @@ def _create_cluster(self) -> Any:
             load_balancing_policy=load_balancing_policy,
             default_retry_policy=retry_policy,
             shard_aware_options=shard_aware_options,
+            address_translator=address_translator,
         )
 
         # Configure connection pool settings
         if cluster.profile_manager is not None:
             profile = cluster.profile_manager.default
             profile.request_timeout = self.config.REQUEST_TIMEOUT
+            # Configure connection pool limits per host
+            profile.max_connections_per_host = self.config.MAX_CONNECTIONS_PER_HOST
+            profile.min_connections_per_host = self.config.MIN_CONNECTIONS_PER_HOST
+            profile.core_connections_per_host = self.config.CORE_CONNECTIONS_PER_HOST
 
         # Set pool configuration
         cluster.connection_class.max_requests_per_connection = self.config.MAX_REQUESTS_PER_CONNECTION
@@ -328,10 +351,23 @@ def create_keyspace(self, keyspace: str, replication_factor: int = 1) -> None:
             keyspace (str): The name of the keyspace to create.
             replication_factor (int): The replication factor. Defaults to 1.
         """
-        query = f"""
-            CREATE KEYSPACE IF NOT EXISTS {keyspace}
-            WITH replication = {{'class': 'SimpleStrategy', 'replication_factor': {replication_factor}}}
-        """
+        # Use configured replication strategy
+        if self.config.REPLICATION_STRATEGY == "NetworkTopologyStrategy" and self.config.REPLICATION_CONFIG:
+            # Build replication config for NetworkTopologyStrategy
+            replication_parts = ["'class': 'NetworkTopologyStrategy'"]
+            for dc, rf in self.config.REPLICATION_CONFIG.items():
+                replication_parts.append(f"'{dc}': {rf}")
+            replication_str = ", ".join(replication_parts)
+            query = f"""
+                CREATE KEYSPACE IF NOT EXISTS {keyspace}
+                WITH replication = {{{replication_str}}}
+            """
+        else:
+            # Use SimpleStrategy (default)
+            query = f"""
+                CREATE KEYSPACE IF NOT EXISTS {keyspace}
+                WITH replication = {{'class': 'SimpleStrategy', 'replication_factor': {replication_factor}}}
+            """
         try:
             self.execute(query)
         except Exception as e:
@@ -765,13 +801,10 @@ def _create_cluster(self) -> Any:
 
         # Configure load balancing policy with optional datacenter awareness
         if self.config.LOCAL_DC:
-            from cassandra.policies import DCAwareRoundRobinPolicy
-
             base_policy = DCAwareRoundRobinPolicy(local_dc=self.config.LOCAL_DC)
+            load_balancing_policy = TokenAwarePolicy(base_policy)
         else:
-            base_policy = RoundRobinPolicy()
-
-        load_balancing_policy = TokenAwarePolicy(base_policy)
+            load_balancing_policy = TokenAwarePolicy(RoundRobinPolicy())
 
         if self.config.RETRY_POLICY == "FALLTHROUGH":
             retry_policy = FallthroughRetryPolicy()
@@ -786,6 +819,12 @@ def _create_cluster(self) -> Any:
         if self.config.DISABLE_SHARD_AWARENESS:
             shard_aware_options = {"disable": True}
 
+        # Address translation for Docker/NAT environments where gossip-discovered
+        # internal container IPs are unreachable from the host
+        address_translator = None
+        if self.config.ADDRESS_TRANSLATION_ENABLED:
+            address_translator = _FixedAddressTranslator(self.config.CONTACT_POINTS[0])
+
         # Cluster is from cassandra.cluster, properly typed
         cluster = Cluster(
             contact_points=self.config.CONTACT_POINTS,
@@ -797,12 +836,17 @@ def _create_cluster(self) -> Any:
             load_balancing_policy=load_balancing_policy,
             default_retry_policy=retry_policy,
             shard_aware_options=shard_aware_options,
+            address_translator=address_translator,
         )
 
         # Configure connection pool settings
         if cluster.profile_manager is not None:
             profile = cluster.profile_manager.default
             profile.request_timeout = self.config.REQUEST_TIMEOUT
+            # Configure connection pool limits per host
+            profile.max_connections_per_host = self.config.MAX_CONNECTIONS_PER_HOST
+            profile.min_connections_per_host = self.config.MIN_CONNECTIONS_PER_HOST
+            profile.core_connections_per_host = self.config.CORE_CONNECTIONS_PER_HOST
 
         # Set pool configuration
         cluster.connection_class.max_requests_per_connection = self.config.MAX_REQUESTS_PER_CONNECTION
@@ -925,10 +969,23 @@ async def create_keyspace(self, keyspace: str, replication_factor: int = 1) -> N
             keyspace (str): The name of the keyspace to create.
             replication_factor (int): The replication factor. Defaults to 1.
         """
-        query = f"""
-            CREATE KEYSPACE IF NOT EXISTS {keyspace}
-            WITH replication = {{'class': 'SimpleStrategy', 'replication_factor': {replication_factor}}}
-        """
+        # Use configured replication strategy
+        if self.config.REPLICATION_STRATEGY == "NetworkTopologyStrategy" and self.config.REPLICATION_CONFIG:
+            # Build replication config for NetworkTopologyStrategy
+            replication_parts = ["'class': 'NetworkTopologyStrategy'"]
+            for dc, rf in self.config.REPLICATION_CONFIG.items():
+                replication_parts.append(f"'{dc}': {rf}")
+            replication_str = ", ".join(replication_parts)
+            query = f"""
+                CREATE KEYSPACE IF NOT EXISTS {keyspace}
+                WITH replication = {{{replication_str}}}
+            """
+        else:
+            # Use SimpleStrategy (default)
+            query = f"""
+                CREATE KEYSPACE IF NOT EXISTS {keyspace}
+                WITH replication = {{'class': 'SimpleStrategy', 'replication_factor': {replication_factor}}}
+            """
         try:
             await self.execute(query)
         except Exception as e:

archipy/configs/config_template.py

@@ -1002,6 +1002,13 @@ class ScyllaDBConfig(BaseModel):
         default=False,
         description="Disable shard awareness (useful for Docker/Testcontainer/NAT environments)",
     )
+    ADDRESS_TRANSLATION_ENABLED: bool = Field(
+        default=False,
+        description="Enable address translation to redirect all discovered node connections to the first contact point. "
+        "In Docker/Testcontainer/NAT environments, ScyllaDB nodes advertise their internal container IPs "
+        "via gossip, which are unreachable from the host. When enabled, the driver translates all discovered "
+        "addresses to the first configured contact point, allowing connections through Docker's port mapping.",
+    )
     RETRY_POLICY: Literal["EXPONENTIAL_BACKOFF", "FALLTHROUGH"] = Field(
         default="EXPONENTIAL_BACKOFF",
         description="Retry policy type (uses native driver RetryPolicy). "

features/steps/kafka_adapter_steps.py

@@ -232,7 +232,7 @@ def step_topic_list_includes(context, topic_name):
 def step_consumer_receive(context, expected_message, topic_name, group_id):
     adapter = get_kafka_consumer_adapter(context, topic_name, group_id)
     try:
-        messages = adapter.batch_consume(messages_number=1, timeout=2)
+        messages = adapter.batch_consume(messages_number=1, timeout=10)
         assert len(messages) > 0, "No messages received"
         received_message = messages[0].value().decode("utf-8")
         assert received_message == expected_message, f"Expected '{expected_message}', got '{received_message}'"

features/test_containers.py

@@ -1,6 +1,7 @@
 """Container manager for test containers"""
 
 import logging
+from pathlib import Path
 
 from testcontainers.core.container import DockerContainer
 from testcontainers.core.waiting_utils import wait_for_logs
@@ -452,8 +453,8 @@ def __init__(self, config: KafkaConfig | None = None, image: str | None = None)
         self.port: int | None = None
         self.bootstrap_servers: str | None = None
 
-        # Set up the container
-        self._container = KafkaContainer(image=self.image)
+        # Set up the container with KRaft mode (required for cp-kafka 7.4+, mandatory for 8.0+)
+        self._container = KafkaContainer(image=self.image).with_kraft()
 
     def start(self) -> KafkaContainer:
         """Start the Kafka container."""
@@ -566,6 +567,9 @@ def stop(self) -> None:
 class ScyllaDBTestContainer(metaclass=Singleton, thread_safe=True):
     """Test container for ScyllaDB."""
 
+    # Minimum aio-max-nr required by ScyllaDB's Seastar framework
+    MIN_AIO_MAX_NR = 131072
+
     def __init__(self, config: ScyllaDBConfig | None = None, image: str | None = None) -> None:
         """Initialize ScyllaDB test container.
 
@@ -597,15 +601,47 @@ def __init__(self, config: ScyllaDBConfig | None = None, image: str | None = Non
         # Add environment variables for single-node configuration
         self._container.with_env("SCYLLA_ARGS", "--smp 1 --memory 750M")
 
+    @staticmethod
+    def _check_aio_max_nr() -> None:
+        """Check that the host system's fs.aio-max-nr is sufficient for ScyllaDB.
+
+        ScyllaDB's Seastar framework requires a minimum number of AIO slots.
+        If the system value is too low, ScyllaDB will crash on startup.
+
+        Raises:
+            RuntimeError: If aio-max-nr is below the required minimum.
+        """
+        aio_path = Path("/proc/sys/fs/aio-max-nr")
+        if not aio_path.exists():
+            logger.warning("Cannot check aio-max-nr: %s not found (non-Linux system?)", aio_path)
+            return
+
+        current_value = int(aio_path.read_text().strip())
+        if current_value < ScyllaDBTestContainer.MIN_AIO_MAX_NR:
+            msg = (
+                f"ScyllaDB requires fs.aio-max-nr >= {ScyllaDBTestContainer.MIN_AIO_MAX_NR}, "
+                f"but the current value is {current_value}. "
+                f"Fix this by running: sudo sysctl -w fs.aio-max-nr={ScyllaDBTestContainer.MIN_AIO_MAX_NR} "
+                f"To make it permanent, add 'fs.aio-max-nr = {ScyllaDBTestContainer.MIN_AIO_MAX_NR}' "
+                f"to /etc/sysctl.conf and run 'sudo sysctl -p'."
+            )
+            raise RuntimeError(msg)
+
     def start(self) -> DockerContainer:
         """Start the ScyllaDB container.
 
         Returns:
             DockerContainer: The running container instance.
+
+        Raises:
+            RuntimeError: If the host system's aio-max-nr is too low for ScyllaDB.
         """
         if self._is_running:
             return self._container
 
+        # Pre-flight check: verify host AIO configuration
+        self._check_aio_max_nr()
+
         # Start the container
         self._container.start()
 

pyproject.toml

@@ -20,25 +20,25 @@ license = { file = "LICENSE" }
 # Core optional dependencies
 aiosqlite = ["aiosqlite>=0.22.1"]
 behave = ["behave>=1.3.3"]
-cache = ["cachetools>=7.0.0", "async-lru>=2.1.0"]
+cache = ["cachetools>=7.0.1", "async-lru>=2.1.0"]
 dependency-injection = ["dependency-injector>=4.48.3"]
 elastic-apm = ["elastic-apm>=6.25.0"]
 elasticsearch = ["elasticsearch>=9.3.0"]
 elasticsearch-async = ["elasticsearch[async]>=9.3.0"]
 fakeredis = ["fakeredis>=2.33.0"]
-fastapi = ["fastapi[all]>=0.128.5"]
+fastapi = ["fastapi[all]>=0.129.0"]
 grpc = ["grpcio>=1.78.0", "grpcio-health-checking>=1.78.0", "protobuf>=6.33.5"]
 jwt = ["pyjwt>=2.11.0"]
 kafka = ["confluent-kafka>=2.13.0"]
 kavenegar = ["kavenegar>=1.1.2"]
-keycloak = ["python-keycloak>=7.0.3", "cachetools>=7.0.0", "async-lru>=2.1.0"]
-minio = ["minio>=7.2.20", "cachetools>=7.0.0", "async-lru>=2.1.0"]
+keycloak = ["python-keycloak>=7.1.1", "cachetools>=7.0.1", "async-lru>=2.1.0"]
+minio = ["minio>=7.2.20", "cachetools>=7.0.1", "async-lru>=2.1.0"]
 parsian-ipg = ["zeep>=4.3.2", "requests[socks]>=2.32.5"]
 postgres = ["psycopg[binary,pool]>=3.3.2"]
 prometheus = ["prometheus-client>=0.24.1"]
-redis = ["redis[hiredis]>=7.1.0"]
+redis = ["redis[hiredis]>=7.1.1"]
 scheduler = ["apscheduler>=3.11.2"]
-scylladb = ["scylla-driver>=3.29.7", "lz4>=4.4.5", "cachetools>=7.0.0", "async-lru>=2.1.0"]
+scylladb = ["scylla-driver>=3.29.8", "lz4>=4.4.5", "cachetools>=7.0.1", "async-lru>=2.1.0"]
 sentry = ["sentry-sdk>=2.52.0"]
 sqlalchemy = ["sqlalchemy>=2.0.46"]
 sqlalchemy-async = ["sqlalchemy[asyncio]>=2.0.46"]
@@ -68,10 +68,10 @@ dev = [
     "add-trailing-comma>=4.0.0",
     "bandit>=1.9.3",
     "codespell>=2.4.1",
-    "ty>=0.0.15",
+    "ty>=0.0.17",
     "pre-commit-hooks>=6.0.0",
     "pre-commit>=4.5.1",
-    "ruff>=0.15.0",
+    "ruff>=0.15.1",
     "types-cachetools>=6.2.0.20251022",
     "types-grpcio>=1.0.0.20251009",
     "types-protobuf>=6.32.1.20251210",