Copilot commented on code in PR #8517:
URL: https://github.com/apache/hbase/pull/8517#discussion_r3918687423
##########
dev-support/read-replica/cluster1/conf/zoo.cfg:
##########
@@ -0,0 +1,3 @@
+clientPort=2181
+clientPortAddress=hbase-docker
+server.1=hbase-docker:2181
Review Comment:
ZooKeeper 3.8 expects `server.<id>=host:quorumPort:electionPort`; this
two-field value is malformed and `QuorumPeerConfig.parseProperties` will reject
it before HBase can start. Use separate quorum and election ports (the standard
defaults are 2888 and 3888).
##########
dev-support/read-replica/cluster2/conf/zoo.cfg:
##########
@@ -0,0 +1,3 @@
+clientPort=2181
+clientPortAddress=hbase-docker-2
+server.1=hbase-docker-2:2181
Review Comment:
ZooKeeper 3.8 expects `server.<id>=host:quorumPort:electionPort`; this
two-field value is malformed and `QuorumPeerConfig.parseProperties` will reject
it before HBase can start. Use separate quorum and election ports (the standard
defaults are 2888 and 3888).
##########
dev-support/read-replica/Dockerfile:
##########
@@ -0,0 +1,131 @@
+# Stage 0: Cache Maven dependencies
+ARG BASE_IMAGE=registry.access.redhat.com/ubi8/openjdk-17:1.23
+FROM ${BASE_IMAGE} AS cache-stage
+
+# Switch to user root (UID 0)
+# hadolint ignore=DL3002
+USER 0
+
+# Install necessary packages for building Maven dependencies
+# hadolint ignore=DL3041
+RUN microdnf update -y && microdnf install -y \
+ maven \
+ git \
+ hostname \
+ diffutils \
+ && microdnf clean all
+
+# Copy the entire source code to cache dependencies
+COPY ./hbase /opt/hbase-src
+
+WORKDIR /opt/hbase-src
+
+# Download and cache all dependencies
+RUN mvn clean install -DskipTests -Dskip.license.check=true
+
+# Stage 1: Build the HBase source code
+FROM ${BASE_IMAGE} AS build-stage
+
+# Switch to user root (UID 0)
+# hadolint ignore=DL3002
+USER 0
+
+# Install necessary build packages
+# hadolint ignore=DL3041
+RUN microdnf update -y && microdnf install -y \
+ maven \
+ git \
+ hostname \
+ diffutils \
+ && microdnf clean all
+
+# Copy the cached Maven dependencies
+COPY --from=cache-stage /root/.m2 /root/.m2
+
+# Copy the HBase source code
+COPY ./hbase /opt/hbase-src
+
+WORKDIR /opt/hbase-src
+
+# Build HBase source code using cached dependencies and enable parallel build
+RUN mvn clean package -DskipTests -Dskip.license.check=true assembly:single -T
1C
+
+# Stage 2: Create the final Docker image
+FROM ${BASE_IMAGE}
+
+# Switch to user root (UID 0)
+USER 0
+
+# Set environment variables
+ENV HBASE_HOME=/opt/hbase
+ENV JAVA_HOME=/usr/lib/jvm/java-17-openjdk \
+ HBASE_USER=hbase \
+ HBASE_CONF_DIR=${HBASE_HOME}/conf \
+ HBASE_LIB_DIR=${HBASE_HOME}/lib \
+ HBASE_LOGS_DIR=${HBASE_HOME}/logs \
+ DATA_DIR=/data-store
+
+# Install necessary runtime packages
+# hadolint ignore=DL3041
+RUN microdnf update -y && microdnf install -y \
+ unzip \
+ gzip \
+ wget \
+ hostname \
+ maven \
+ git \
+ diffutils \
+ vim \
+ openssh-clients \
+ python3 \
+ procps \
+ && microdnf clean all
+
+# Copy the built HBase binaries from the build-stage
+COPY --from=build-stage
/opt/hbase-src/hbase-assembly/target/hbase-4.0.0-alpha-1-SNAPSHOT-bin.tar.gz
/opt/
+
+# Extract HBase binaries
+RUN tar -xzf /opt/hbase-4.0.0-alpha-1-SNAPSHOT-bin.tar.gz -C /opt \
+ && ln -s /opt/hbase-4.0.0-alpha-1-SNAPSHOT /opt/hbase \
+ && rm /opt/hbase-4.0.0-alpha-1-SNAPSHOT-bin.tar.gz
Review Comment:
This hard-coded master version contradicts the Jenkins stage's `branch-3`
condition: branch-3 currently has revision `3.1.0-SNAPSHOT`, while the assembly
name is derived from `${project.version}` (`hbase-assembly/pom.xml:35`). That
build cannot find this source archive, and the nightly will also break at the
next master version bump. Select the non-client assembly archive without
embedding a version.
##########
dev-support/hbase_nightly_read_replica_test.sh:
##########
@@ -0,0 +1,117 @@
+#!/usr/bin/env bash
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+# Run the read-replica Docker integration test suite.
+
+set -e
+
+SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
+REPLICA_DIR="${SCRIPT_DIR}/read-replica"
+export HBASE_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)"
+
+export HBASE_IMAGE="hbase-read-replica:${BUILD_NUMBER:-local}"
+
+echo "Script dir: ${SCRIPT_DIR}"
+echo "Replica dir: ${REPLICA_DIR}"
+echo "HBase root: ${HBASE_ROOT}"
+
+echo "Changing to replica dir: REPLICA_DIR"
+cd "${REPLICA_DIR}"
+
+echo "Sourcing environment file: $(pwd)/.env"
+set -a
+source .env
+set +a
+
+echo "HBASE_IMAGE=${HBASE_IMAGE}"
+echo "ACTIVE_CLUSTER_CONF_DIR=${ACTIVE_CLUSTER_CONF_DIR}"
+echo "REPLICA_CLUSTER_CONF_DIR=${REPLICA_CLUSTER_CONF_DIR}"
+echo "DOCKER_COMPOSE_FILE=${DOCKER_COMPOSE_FILE}"
+echo "HBASE_DATA_STORE_ROOT=${HBASE_DATA_STORE_ROOT}"
+echo "realpath of HBASE_DATA_STORE_ROOT=$(realpath ${HBASE_DATA_STORE_ROOT})"
+
+echo "Removing HBase log directories from mounted volumes that may exist from
a previous test run:"
+echo "ACTIVE_CLUSTER_LOGS_DIR=${ACTIVE_CLUSTER_LOGS_DIR}"
+echo "REPLICA_CLUSTER_LOGS_DIR=${REPLICA_CLUSTER_LOGS_DIR}"
+rm -rf ${ACTIVE_CLUSTER_LOGS_DIR} ${REPLICA_CLUSTER_LOGS_DIR}
Review Comment:
These directories are immediately used as bind-mount sources, but after
deletion a rootful Docker daemon recreates missing host directories as
root-owned and non-writable by the image's UID 1000. HBase can then fail to
create its log files and the containers exit. Recreate the directories with
writable permissions before `docker compose up`.
##########
dev-support/read-replica/python/scripts/test_dual_active_cluster_startup.py:
##########
@@ -0,0 +1,140 @@
+#!/usr/bin/env python3
+"""
+Verifies that two clusters cannot both start with read-only mode disabled
(both as active clusters)
+on the same shared data store. One cluster must fail to start, with the
HMaster process not
+running, and an error logged to the master log.
+
+Usage: python3 ./python/scripts/test_dual_active_cluster_startup.py
+"""
+import argparse
+import os
+import time
+
+from python.src.environment_loader import get_env
+from python.src.hbase_docker_client import HBaseDockerClient
+from python.src.logger_config import get_logger
+from python.src.utils import load_env_and_set_up_clients, log_script_start,
log_script_end
+
+logger = get_logger(__name__)
+
+STARTUP_WAIT_SECONDS = 60
+EXPECTED_ERROR_MSG = "Another cluster is running in active (read-write) mode
on this storage location"
+CLUSTER1_SERVICE_NAME = "hbase"
+CLUSTER2_SERVICE_NAME = "hbase2"
+
+
+def is_process_running(cluster: HBaseDockerClient, process_name: str) -> bool:
+ output = cluster.run_docker_exec_command("jps")
+ return process_name in output
+
+
+def check_cluster_processes(cluster: HBaseDockerClient) -> bool:
+ hmaster_running = is_process_running(cluster, "HMaster")
+ logger.info(f" {cluster.name}: HMaster={'running' if hmaster_running else
'down'}")
+ return hmaster_running
+
+
+def assert_error_in_master_log(cluster: HBaseDockerClient):
+ logger.info(f"Checking {cluster.name} master log for expected error
message")
+ log_output = cluster.run_docker_exec_command(
+ "cat /opt/hbase/logs/hbase-*-master-*.log || true"
Review Comment:
This reads every master log retained in the mounted directory across all
four iterations. When a cluster is the second starter again, an expected error
from its earlier iteration remains in this output, so an unrelated current
startup crash can satisfy the assertion. Restrict the check to bytes/log files
produced after the current start attempt, or clear that stopped cluster's logs
before starting it.
##########
dev-support/read-replica/python/src/hbase_docker_client.py:
##########
@@ -0,0 +1,623 @@
+#!/usr/bin/env python3
+import ast
+import logging
+import re
+from concurrent.futures import ThreadPoolExecutor, TimeoutError as
FuturesTimeoutError
+
+import docker
+import requests
+import subprocess
+import time
+import xml.etree.ElementTree as ET
+
+from .logger_config import get_logger
+
+logger = get_logger(__name__)
+
+
+class DockerExecCommandError(Exception):
+ pass
+
+
+class HBaseShellCommandError(DockerExecCommandError):
+ pass
+
+
+class DockerExecCommandTimeoutError(DockerExecCommandError):
+ pass
+
+
+class HBaseDockerClient:
+ def __init__(self, container_name: str, local_conf: str, hbase_ui_port:
int = 16010,
+ cluster_name: str = "HBase Cluster", max_retries: int = 12,
sleep_time: int = 5,
+ hbase_host: str = "localhost") -> None:
+ self._container_name = container_name
+ self._local_conf = local_conf
+ self._hbase_ui_port = hbase_ui_port
+ self._cluster_name = cluster_name
+ self._max_retries = max_retries
+ self._sleep_time = sleep_time
+ self._hbase_host = hbase_host
+ self._docker_client = docker.from_env()
+
+ @property
+ def name(self) -> str:
+ return self._cluster_name
+
+ def run_docker_exec_command(self, bash_cmd: str, timeout: int | None =
None) -> str:
+ """
+ Uses the Docker SDK to exec a Bash command in the object's Docker
container.
+ Equivalent to: docker exec <container> bash -c <bash_cmd>
+ """
+ cmd = ["bash", "-c", bash_cmd]
+ cmd_str = f"docker exec {self._container_name} bash -c {bash_cmd}"
+ logger.debug(f"Running command on {self._cluster_name}: {cmd_str}")
+
+ try:
+ container =
self._docker_client.containers.get(self._container_name)
+
+ if timeout is not None:
+ with ThreadPoolExecutor(max_workers=1) as pool:
+ future = pool.submit(container.exec_run, cmd, demux=True)
+ try:
+ result = future.result(timeout=timeout)
+ except FuturesTimeoutError:
+ raise DockerExecCommandTimeoutError(
+ f"Command timed out after {timeout}s on
{self._cluster_name} "
+ f"({self._container_name}): {bash_cmd}\n"
+ f"The command used to run this was: {cmd_str}\n"
+ )
+ else:
+ result = container.exec_run(cmd, demux=True)
+ except DockerExecCommandError:
+ raise
+ except docker.errors.DockerException as e:
+ raise DockerExecCommandError(
+ f"The following command failed on {self._cluster_name}
({self._container_name}): {bash_cmd}\n"
+ f"The command used to run this was: {cmd_str}\n"
+ f"Docker error: {e}\n"
+ )
+
+ exit_code, (stdout, stderr) = result
+ stdout_str = (stdout or b'').decode('utf-8')
+ if exit_code != 0:
+ raise DockerExecCommandError(
+ f"The following command failed on {self._cluster_name}
({self._container_name}): {bash_cmd}\n"
+ f"The command used to run this was: {cmd_str}\n"
+ f"The command's STDERR was:\n{(stderr or
b'').decode('utf-8')}\n"
+ f"The command's STDOUT was:\n{stdout_str}\n"
+ )
+ return stdout_str
+
+ def run_hbase_shell_command(self, hbase_cmd: str, timeout: int | None =
None) -> str:
+ """
+ Uses 'docker exec' to run the provided HBase shell command in the
object's Docker container.
+ The command looks like: docker exec <container> bash -c hbase shell -n
<<< "<hbase_cmd>"
+ """
+ hbase_shell_cmd = f'''hbase shell -n <<< "{hbase_cmd}"'''
+ try:
+ return self.run_docker_exec_command(hbase_shell_cmd,
timeout=timeout)
+ except DockerExecCommandTimeoutError:
+ # DockerExecCommandTimeoutError is a subclass of
DockerExecCommandError, so we need to make sure
+ # it's specifically caught and re-raised. Otherwise, it's
swallowed when catching DockerExecCommandError
+ raise
+ except DockerExecCommandError as e:
+ raise HBaseShellCommandError(e)
+
+ def _get_pid_from_jps(self, process_name: str) -> int | None:
+ """Runs jps inside the container and returns the PID of the named
process, or None."""
+ try:
+ output = self.run_docker_exec_command("jps")
+ for line in output.strip().splitlines():
+ parts = line.split()
+ if len(parts) == 2 and parts[1] == process_name:
+ return int(parts[0])
+ except DockerExecCommandError:
+ pass
+ return None
+
+ def wait_for_hbase_ui(self) -> bool:
+ """Checks for a 200 OK on the HBase Master UI."""
+ # Read HBASE_HOST from environment, falling back to 'localhost' for
host-native execution
+ url = f"http://{self._hbase_host}:{self._hbase_ui_port}";
+ logger.info(f"Waiting for HBase UI: {self._cluster_name} on {url}")
+ last_exception = None
+ for attempt in range(1, self._max_retries + 1):
+ try:
+ response = requests.get(url)
+ if response.status_code == 200:
+ logger.info(f"SUCCESS: {self._cluster_name} UI is up.")
+ return True
+ except requests.exceptions.ConnectionError as e:
+ last_exception = e
+ logging.info(f"Waiting {self._sleep_time} seconds before
requesting HBase UI again")
+ time.sleep(self._sleep_time)
+
+ raise RuntimeError(f"\nTIMEOUT: {self._cluster_name} UI failed to
respond after "
+ f"{self._max_retries} attempts. "
+ f"Last raised exception was: {last_exception}")
+
+ def wait_for_master_initialization(self) -> bool:
+ """Waits for the current HMaster process to log 'Master has completed
initialization'."""
+ logger.info(f"Waiting for Master initialization: {self._cluster_name}
({self._container_name})")
+ for attempt in range(1, self._max_retries + 1):
+ pid = self._get_pid_from_jps("HMaster")
+ if pid is not None:
+ awk_cmd = (
+ f"awk '/env:JVM_PID={pid}/{{seen=1; found=0}} "
+ f"seen && /Master has completed initialization/{{found=1}}
"
+ f"END{{exit !found}}' /opt/hbase/logs/hbase-*-master-*.log"
+ )
+ try:
+ self.run_docker_exec_command(awk_cmd)
+ logger.info(f"SUCCESS: {self._cluster_name} Master has
completed initialization.")
+ return True
+ except DockerExecCommandError:
+ pass
+ logging.info(f"Waiting {self._sleep_time} seconds before checking
Master initialization again")
+ time.sleep(self._sleep_time)
+
+ raise RuntimeError(
+ f"\nTIMEOUT: {self._cluster_name} Master failed to initialize
after "
+ f"{self._max_retries} attempts.")
+
+ def wait_for_region_server_initialization(self) -> bool:
+ """Waits for the current HRegionServer process to log 'Serving as'
message."""
+ logger.info(f"Waiting for RegionServer initialization:
{self._cluster_name} ({self._container_name})")
+ for attempt in range(1, self._max_retries + 1):
+ pid = self._get_pid_from_jps("HRegionServer")
+ if pid is not None:
+ awk_cmd = (
+ f"awk '/env:JVM_PID={pid}/{{seen=1; found=0}} "
+ f"seen && /Serving as {self._container_name},/{{found=1}} "
+ f"END{{exit !found}}'
/opt/hbase/logs/hbase-*-regionserver-*.log"
+ )
+ try:
+ self.run_docker_exec_command(awk_cmd)
+ logger.info(f"SUCCESS: {self._cluster_name} RegionServer
is serving.")
+ return True
+ except DockerExecCommandError:
+ pass
+ logging.info(f"Waiting {self._sleep_time} seconds before checking
RegionServer initialization again")
+ time.sleep(self._sleep_time)
+
+ raise RuntimeError(
+ f"\nTIMEOUT: {self._cluster_name} RegionServer failed to
initialize after "
+ f"{self._max_retries} attempts.")
+
+ def check_server_status(self, desired_status: dict | None = None) -> bool:
+ """Runs 'status' inside the HBase shell and validates the output."""
+ if desired_status is None:
+ desired_status = {'masters': '1', 'region_servers': '1',
'dead_servers': '0'}
+ logger.info(f"Validating Cluster Status: {self._cluster_name}
({self._container_name})")
+ for attempt in range(1, self._max_retries + 1):
+ try:
+ output = self.get_hbase_status()
+
+ # The cluster's status should have 1 active master, 1 region
server,
+ # and no dead servers
+ validations = {
+ "Active Master": f"{desired_status['masters']} active
master" in output,
+ "Region Server": f"{desired_status['region_servers']}
servers" in output,
+ "No Dead Servers": f"{desired_status['dead_servers']}
dead" in output
+ }
+
+ if all(validations.values()):
+ for check, status in validations.items():
+ logger.info(f" [PASS] {check}")
+ logger.info(f"SUCCESS: {self._cluster_name} is fully
operational.")
+ return True
+ else:
+ logger.warning(f"{self._cluster_name} is responding, but
not all "
+ f"components are ready...")
+ logger.info(f"HBase 'status' command output:\n{output}")
+
+ except HBaseShellCommandError:
+ pass
+
+ logging.info(f"Waiting {self._sleep_time} seconds before getting
status on {self.name} again")
+ time.sleep(self._sleep_time)
+
+ raise RuntimeError(
+ f"\nTIMEOUT: {self._cluster_name} shell check failed after
{self._max_retries} attempts.")
+
+ def get_hbase_status(self) -> str:
+ logger.debug(f"Getting status of {self.name}")
+ return self.run_hbase_shell_command("status")
+
+ def wait_for_cluster_to_start(self) -> None:
+ """curls the cluster's HBase UI to make sure it is up and then makes
sure all desired servers are up"""
+ self.wait_for_hbase_ui()
+ self.wait_for_master_initialization()
+ self.wait_for_region_server_initialization()
+ self.check_server_status()
+
+ def create_table(self, table_name: str, column_family: str) -> bool:
+ logger.info(f"Creating table '{table_name}' on {self._cluster_name}")
+ create_cmd = f"create '{table_name}', '{column_family}'"
+ output = self.run_hbase_shell_command(create_cmd)
+
+ if f"Created table {table_name}" not in output:
+ logger.error(f"Could not create table '{table_name}' on
{self._cluster_name}")
+ return False
+ return True
+
+ def disable_table(self, table_name: str) -> None:
+ logger.debug(f"Disabling table '{table_name}' on {self.name}")
+ self.run_hbase_shell_command(f"disable '{table_name}'")
+
+ def drop_table(self, table_name: str) -> None:
+ logger.info(f"Dropping table '{table_name}' on {self.name}")
+ self.run_hbase_shell_command(f"drop '{table_name}'")
+
+ def list_tables(self) -> list:
+ """Gets the list of HBase tables and returns it as a Python list"""
+ logger.debug(f"Getting the list of tables in HBase on {self.name}")
+ pattern = r'\[(.*?)\]'
+ output = self.run_hbase_shell_command("list")
+ output = output.replace('\n', ' ')
+ match = re.search(pattern, output)
+ return ast.literal_eval(match.group(0))
+
+ def list_regions(self, table_name: str) -> str:
+ """Gets list of regions and their info for the provided table"""
+ logger.info(f"Getting list of regions for table '{table_name}'")
+ return self.run_hbase_shell_command(f"list_regions '{table_name}'")
+
+ def put(self, table_name: str, row: str, column: str, data: str, spec_map:
str | None = None) -> None:
+ """
+ Performs an HBase put command.
+ :param table_name: the table we are inserting data into
+ :param row: row of the table we are inserting data into
+ :param column: column of the table we are inserting data into
+ :param data: the actual data we are inserting (as a string)
+ :param spec_map: additional attributes input as a string
+ (e.g. "{ATTRIBUTES=>{'my-key'=>'my-value'}}")
+ """
+ logger.info(f"Adding data to table '{table_name}' on {self.name}")
+ put_cmd = f"put '{table_name}', '{row}', '{column}', '{data}'"
+ if spec_map:
+ put_cmd += f", {spec_map}"
+ self.run_hbase_shell_command(put_cmd)
+
+ def get(self, table_name: str, row: str, column: str | None = None,
spec_map: str | None = None) -> str:
+ logger.info(f"Getting data from table '{table_name}' on {self.name}")
+ get_cmd = f"get '{table_name}', '{row}'"
+ if column:
+ get_cmd += f", '{column}'"
+ if spec_map:
+ get_cmd += f", {spec_map}"
+ output = self.run_hbase_shell_command(get_cmd)
+ logger.debug(f"Got data:\n{output}")
+ return output
+
+ def delete(self, table_name: str, row: str, column: str, timestamp: int |
None = None,
+ spec_map: str | None = None) -> None:
+ logger.info(f"Deleting data from table '{table_name}' on {self.name}")
+ delete_cmd = f"delete '{table_name}', '{row}', '{column}'"
+ if timestamp:
+ delete_cmd += f", {table_name}"
Review Comment:
When a timestamp is supplied, this appends the table name instead of the
timestamp, producing an invalid HBase shell command such as `..., my_table`;
timestamp `0` is also silently omitted by the truthiness check. The optional
timestamp API therefore does not construct the requested delete.
##########
dev-support/read-replica/python/src/hbase_docker_client.py:
##########
@@ -0,0 +1,623 @@
+#!/usr/bin/env python3
+import ast
+import logging
+import re
+from concurrent.futures import ThreadPoolExecutor, TimeoutError as
FuturesTimeoutError
+
+import docker
+import requests
+import subprocess
+import time
+import xml.etree.ElementTree as ET
+
+from .logger_config import get_logger
+
+logger = get_logger(__name__)
+
+
+class DockerExecCommandError(Exception):
+ pass
+
+
+class HBaseShellCommandError(DockerExecCommandError):
+ pass
+
+
+class DockerExecCommandTimeoutError(DockerExecCommandError):
+ pass
+
+
+class HBaseDockerClient:
+ def __init__(self, container_name: str, local_conf: str, hbase_ui_port:
int = 16010,
+ cluster_name: str = "HBase Cluster", max_retries: int = 12,
sleep_time: int = 5,
+ hbase_host: str = "localhost") -> None:
+ self._container_name = container_name
+ self._local_conf = local_conf
+ self._hbase_ui_port = hbase_ui_port
+ self._cluster_name = cluster_name
+ self._max_retries = max_retries
+ self._sleep_time = sleep_time
+ self._hbase_host = hbase_host
+ self._docker_client = docker.from_env()
+
+ @property
+ def name(self) -> str:
+ return self._cluster_name
+
+ def run_docker_exec_command(self, bash_cmd: str, timeout: int | None =
None) -> str:
+ """
+ Uses the Docker SDK to exec a Bash command in the object's Docker
container.
+ Equivalent to: docker exec <container> bash -c <bash_cmd>
+ """
+ cmd = ["bash", "-c", bash_cmd]
+ cmd_str = f"docker exec {self._container_name} bash -c {bash_cmd}"
+ logger.debug(f"Running command on {self._cluster_name}: {cmd_str}")
+
+ try:
+ container =
self._docker_client.containers.get(self._container_name)
+
+ if timeout is not None:
+ with ThreadPoolExecutor(max_workers=1) as pool:
+ future = pool.submit(container.exec_run, cmd, demux=True)
+ try:
+ result = future.result(timeout=timeout)
+ except FuturesTimeoutError:
+ raise DockerExecCommandTimeoutError(
+ f"Command timed out after {timeout}s on
{self._cluster_name} "
+ f"({self._container_name}): {bash_cmd}\n"
+ f"The command used to run this was: {cmd_str}\n"
+ )
+ else:
+ result = container.exec_run(cmd, demux=True)
+ except DockerExecCommandError:
+ raise
+ except docker.errors.DockerException as e:
+ raise DockerExecCommandError(
+ f"The following command failed on {self._cluster_name}
({self._container_name}): {bash_cmd}\n"
+ f"The command used to run this was: {cmd_str}\n"
+ f"Docker error: {e}\n"
+ )
+
+ exit_code, (stdout, stderr) = result
+ stdout_str = (stdout or b'').decode('utf-8')
+ if exit_code != 0:
+ raise DockerExecCommandError(
+ f"The following command failed on {self._cluster_name}
({self._container_name}): {bash_cmd}\n"
+ f"The command used to run this was: {cmd_str}\n"
+ f"The command's STDERR was:\n{(stderr or
b'').decode('utf-8')}\n"
+ f"The command's STDOUT was:\n{stdout_str}\n"
+ )
+ return stdout_str
+
+ def run_hbase_shell_command(self, hbase_cmd: str, timeout: int | None =
None) -> str:
+ """
+ Uses 'docker exec' to run the provided HBase shell command in the
object's Docker container.
+ The command looks like: docker exec <container> bash -c hbase shell -n
<<< "<hbase_cmd>"
+ """
+ hbase_shell_cmd = f'''hbase shell -n <<< "{hbase_cmd}"'''
+ try:
+ return self.run_docker_exec_command(hbase_shell_cmd,
timeout=timeout)
+ except DockerExecCommandTimeoutError:
+ # DockerExecCommandTimeoutError is a subclass of
DockerExecCommandError, so we need to make sure
+ # it's specifically caught and re-raised. Otherwise, it's
swallowed when catching DockerExecCommandError
+ raise
+ except DockerExecCommandError as e:
+ raise HBaseShellCommandError(e)
+
+ def _get_pid_from_jps(self, process_name: str) -> int | None:
+ """Runs jps inside the container and returns the PID of the named
process, or None."""
+ try:
+ output = self.run_docker_exec_command("jps")
+ for line in output.strip().splitlines():
+ parts = line.split()
+ if len(parts) == 2 and parts[1] == process_name:
+ return int(parts[0])
+ except DockerExecCommandError:
+ pass
+ return None
+
+ def wait_for_hbase_ui(self) -> bool:
+ """Checks for a 200 OK on the HBase Master UI."""
+ # Read HBASE_HOST from environment, falling back to 'localhost' for
host-native execution
+ url = f"http://{self._hbase_host}:{self._hbase_ui_port}";
+ logger.info(f"Waiting for HBase UI: {self._cluster_name} on {url}")
+ last_exception = None
+ for attempt in range(1, self._max_retries + 1):
+ try:
+ response = requests.get(url)
+ if response.status_code == 200:
+ logger.info(f"SUCCESS: {self._cluster_name} UI is up.")
+ return True
+ except requests.exceptions.ConnectionError as e:
+ last_exception = e
Review Comment:
Each readiness attempt can block indefinitely because Requests has no
default timeout. If the UI accepts a connection but stalls, retries never
advance and this stage can consume the pipeline's global timeout. Give the
request a bounded timeout and handle all request failures as retryable here.
##########
dev-support/read-replica/python/src/hbase_docker_client.py:
##########
@@ -0,0 +1,623 @@
+#!/usr/bin/env python3
+import ast
+import logging
+import re
+from concurrent.futures import ThreadPoolExecutor, TimeoutError as
FuturesTimeoutError
+
+import docker
+import requests
+import subprocess
+import time
+import xml.etree.ElementTree as ET
+
+from .logger_config import get_logger
+
+logger = get_logger(__name__)
+
+
+class DockerExecCommandError(Exception):
+ pass
+
+
+class HBaseShellCommandError(DockerExecCommandError):
+ pass
+
+
+class DockerExecCommandTimeoutError(DockerExecCommandError):
+ pass
+
+
+class HBaseDockerClient:
+ def __init__(self, container_name: str, local_conf: str, hbase_ui_port:
int = 16010,
+ cluster_name: str = "HBase Cluster", max_retries: int = 12,
sleep_time: int = 5,
+ hbase_host: str = "localhost") -> None:
+ self._container_name = container_name
+ self._local_conf = local_conf
+ self._hbase_ui_port = hbase_ui_port
+ self._cluster_name = cluster_name
+ self._max_retries = max_retries
+ self._sleep_time = sleep_time
+ self._hbase_host = hbase_host
+ self._docker_client = docker.from_env()
+
+ @property
+ def name(self) -> str:
+ return self._cluster_name
+
+ def run_docker_exec_command(self, bash_cmd: str, timeout: int | None =
None) -> str:
+ """
+ Uses the Docker SDK to exec a Bash command in the object's Docker
container.
+ Equivalent to: docker exec <container> bash -c <bash_cmd>
+ """
+ cmd = ["bash", "-c", bash_cmd]
+ cmd_str = f"docker exec {self._container_name} bash -c {bash_cmd}"
+ logger.debug(f"Running command on {self._cluster_name}: {cmd_str}")
+
+ try:
+ container =
self._docker_client.containers.get(self._container_name)
+
+ if timeout is not None:
+ with ThreadPoolExecutor(max_workers=1) as pool:
+ future = pool.submit(container.exec_run, cmd, demux=True)
+ try:
+ result = future.result(timeout=timeout)
+ except FuturesTimeoutError:
+ raise DockerExecCommandTimeoutError(
+ f"Command timed out after {timeout}s on
{self._cluster_name} "
+ f"({self._container_name}): {bash_cmd}\n"
+ f"The command used to run this was: {cmd_str}\n"
+ )
Review Comment:
The timeout is not actually bounded: leaving this `with` block calls
`ThreadPoolExecutor.shutdown(wait=True)`, so after `future.result` times out
the code still waits for `exec_run` to finish before raising. This defeats the
60-second flush guard and can hang the nightly stage until the global 16-hour
timeout. Shut the executor down without waiting on the timeout path.
##########
dev-support/read-replica/Dockerfile:
##########
@@ -0,0 +1,131 @@
+# Stage 0: Cache Maven dependencies
+ARG BASE_IMAGE=registry.access.redhat.com/ubi8/openjdk-17:1.23
+FROM ${BASE_IMAGE} AS cache-stage
+
+# Switch to user root (UID 0)
+# hadolint ignore=DL3002
+USER 0
+
+# Install necessary packages for building Maven dependencies
+# hadolint ignore=DL3041
+RUN microdnf update -y && microdnf install -y \
+ maven \
+ git \
+ hostname \
+ diffutils \
+ && microdnf clean all
+
+# Copy the entire source code to cache dependencies
+COPY ./hbase /opt/hbase-src
Review Comment:
This “cache” layer is keyed on the entire HBase source tree, so every source
change invalidates it and reruns a full `mvn clean install`; the next stage
then performs another full Maven package build. For a nightly that normally
tests a new commit, this provides no dependency-cache benefit and roughly
doubles build work. Use a persistent BuildKit Maven cache in the build stage,
or copy only dependency descriptors before resolving dependencies.
##########
dev-support/read-replica/Dockerfile:
##########
@@ -0,0 +1,131 @@
+# Stage 0: Cache Maven dependencies
Review Comment:
This new file lacks the ASF license header, as do most substantive files
added under `dev-support/read-replica`. The root Apache RAT configuration scans
non-hidden files (`pom.xml:2084-2151`), so release/license checks will reject
these files. Add the standard ASF header to every new substantive Python,
shell, Docker, YAML, requirements, XML, and ZooKeeper configuration file; the
new log4j files and nightly wrapper show the expected header.
##########
dev-support/read-replica/.env:
##########
@@ -0,0 +1,33 @@
+# The name of the HBase Docker image
+HBASE_IMAGE=${HBASE_IMAGE:-kgeisz/hbase-docker:read-replica-jenkins}
+# The name of the HBase docker container
+HBASE_CONTAINER_NAME=hbase-docker
+# This is the host running the hbase-docker containers. Use localhost if the
containers
+# are running locally. If they are started by another container, such as a
Jenkins
+# container in a Docker-out-of-Docker setup, then try setting this to
host.docker.internal.
+HBASE_HOST=localhost
Review Comment:
The preceding comment says callers can set this for Docker-out-of-Docker,
but sourcing `.env` unconditionally overwrites any supplied `HBASE_HOST`. Such
a Jenkins agent will still probe its own localhost and fail readiness checks.
Preserve an existing environment override, as is already done for `HBASE_IMAGE`.
--
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]
