Update the versions for ClickHouse and Debezium (#12741)

## Problem
The test for logical replication used the year-old versions of
ClickHouse and Debezium so that we may miss problems related to
up-to-date versions.
## Summary of changes
The ClickHouse version has been updated to 24.8.
The Debezium version has been updated to the latest stable one,
3.1.3Final.
Some problems with locally running the Debezium test have been fixed.

---------

Co-authored-by: Alexey Masterov <alexey.masterov@databricks.com>
Co-authored-by: Alexander Bayandin <alexander@neon.tech>

.github/workflows/pg-clients.yml

@@ -48,8 +48,20 @@ jobs:
     uses: ./.github/workflows/build-build-tools-image.yml
     secrets: inherit
 
+  generate-ch-tmppw:
+    runs-on: ubuntu-22.04
+    outputs:
+      tmp_val: ${{ steps.pwgen.outputs.tmp_val }}
+    steps:
+      - name: Generate a random password
+        id: pwgen
+        run: |
+          set +x
+          p=$(dd if=/dev/random bs=14 count=1 2>/dev/null | base64)
+          echo tmp_val="${p//\//}" >> "${GITHUB_OUTPUT}"
+
   test-logical-replication:
-    needs: [ build-build-tools-image ]
+    needs: [ build-build-tools-image, generate-ch-tmppw ]
     runs-on: ubuntu-22.04
 
     container:
@@ -60,16 +72,20 @@ jobs:
       options: --init --user root
     services:
       clickhouse:
-        image: clickhouse/clickhouse-server:24.6.3.64
+        image: clickhouse/clickhouse-server:24.8
+        env:
+          CLICKHOUSE_PASSWORD: ${{ needs.generate-ch-tmppw.outputs.tmp_val }}
         ports:
           - 9000:9000
           - 8123:8123
       zookeeper:
-        image: quay.io/debezium/zookeeper:2.7
+        image: quay.io/debezium/zookeeper:3.1.3.Final
         ports:
           - 2181:2181
+          - 2888:2888
+          - 3888:3888
       kafka:
-        image: quay.io/debezium/kafka:2.7
+        image: quay.io/debezium/kafka:3.1.3.Final
         env:
           ZOOKEEPER_CONNECT: "zookeeper:2181"
           KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
@@ -79,7 +95,7 @@ jobs:
         ports:
           - 9092:9092
       debezium:
-        image: quay.io/debezium/connect:2.7
+        image: quay.io/debezium/connect:3.1.3.Final
         env:
           BOOTSTRAP_SERVERS: kafka:9092
           GROUP_ID: 1
@@ -125,6 +141,7 @@ jobs:
           aws-oidc-role-arn: ${{ vars.DEV_AWS_OIDC_ROLE_ARN }}
         env:
           BENCHMARK_CONNSTR: ${{ steps.create-neon-project.outputs.dsn }}
+          CLICKHOUSE_PASSWORD: ${{ needs.generate-ch-tmppw.outputs.tmp_val }}
 
       - name: Delete Neon Project
         if: always()

test_runner/logical_repl/README.md

@@ -9,9 +9,10 @@
 
 ```bash
 export BENCHMARK_CONNSTR=postgres://user:pass@ep-abc-xyz-123.us-east-2.aws.neon.build/neondb
+export CLICKHOUSE_PASSWORD=ch_password123
 
 docker compose -f test_runner/logical_repl/clickhouse/docker-compose.yml up -d
-./scripts/pytest -m remote_cluster -k test_clickhouse
+./scripts/pytest -m remote_cluster -k 'test_clickhouse[release-pg17]'
 docker compose -f test_runner/logical_repl/clickhouse/docker-compose.yml down
 ```
 
@@ -21,6 +22,6 @@ docker compose -f test_runner/logical_repl/clickhouse/docker-compose.yml down
 export BENCHMARK_CONNSTR=postgres://user:pass@ep-abc-xyz-123.us-east-2.aws.neon.build/neondb
 
 docker compose -f test_runner/logical_repl/debezium/docker-compose.yml up -d
-./scripts/pytest -m remote_cluster -k test_debezium
+./scripts/pytest -m remote_cluster -k 'test_debezium[release-pg17]'
 docker compose -f test_runner/logical_repl/debezium/docker-compose.yml down
 ```

test_runner/logical_repl/clickhouse/docker-compose.yml

@@ -1,9 +1,11 @@
 services:
   clickhouse:
-    image: clickhouse/clickhouse-server
+    image: clickhouse/clickhouse-server:25.6
     user: "101:101"
     container_name: clickhouse
     hostname: clickhouse
+    environment:
+      - CLICKHOUSE_PASSWORD=${CLICKHOUSE_PASSWORD:-ch_password123}
     ports:
       - 127.0.0.1:8123:8123
       - 127.0.0.1:9000:9000

test_runner/logical_repl/debezium/docker-compose.yml

@@ -1,18 +1,28 @@
 services:
   zookeeper:
-    image: quay.io/debezium/zookeeper:2.7
+    image: quay.io/debezium/zookeeper:3.1.3.Final
+    ports:
+      - 127.0.0.1:2181:2181
+      - 127.0.0.1:2888:2888
+      - 127.0.0.1:3888:3888
   kafka:
-    image: quay.io/debezium/kafka:2.7
+    image: quay.io/debezium/kafka:3.1.3.Final
+    depends_on: [zookeeper]
     environment:
       ZOOKEEPER_CONNECT: "zookeeper:2181"
-      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
+      KAFKA_LISTENERS: INTERNAL://:9092,EXTERNAL://:29092
+      KAFKA_ADVERTISED_LISTENERS: INTERNAL://kafka:9092,EXTERNAL://localhost:29092
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:PLAINTEXT,EXTERNAL:PLAINTEXT
+      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
       KAFKA_BROKER_ID: 1
       KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
       KAFKA_JMX_PORT: 9991
     ports:
-      - 127.0.0.1:9092:9092
+      - 9092:9092
+      - 29092:29092
   debezium:
-    image: quay.io/debezium/connect:2.7
+    image: quay.io/debezium/connect:3.1.3.Final
+    depends_on: [kafka]
     environment:
       BOOTSTRAP_SERVERS: kafka:9092
       GROUP_ID: 1

test_runner/logical_repl/test_clickhouse.py

@@ -53,8 +53,13 @@ def test_clickhouse(remote_pg: RemotePostgres):
     cur.execute("CREATE TABLE table1 (id integer primary key, column1 varchar(10));")
     cur.execute("INSERT INTO table1 (id, column1) VALUES (1, 'abc'), (2, 'def');")
     conn.commit()
-    client = clickhouse_connect.get_client(host=clickhouse_host)
+    if "CLICKHOUSE_PASSWORD" not in os.environ:
+        raise RuntimeError("CLICKHOUSE_PASSWORD is not set")
+    client = clickhouse_connect.get_client(
+        host=clickhouse_host, password=os.environ["CLICKHOUSE_PASSWORD"]
+    )
     client.command("SET allow_experimental_database_materialized_postgresql=1")
+    client.command("DROP DATABASE IF EXISTS db1_postgres")
     client.command(
         "CREATE DATABASE db1_postgres ENGINE = "
         f"MaterializedPostgreSQL('{conn_options['host']}', "

test_runner/logical_repl/test_debezium.py

@@ -17,6 +17,7 @@ from fixtures.utils import wait_until
 
 if TYPE_CHECKING:
     from fixtures.neon_fixtures import RemotePostgres
+    from kafka import KafkaConsumer
 
 
 class DebeziumAPI:
@@ -101,9 +102,13 @@ def debezium(remote_pg: RemotePostgres):
     assert len(dbz.list_connectors()) == 1
     from kafka import KafkaConsumer
 
+    kafka_host = "kafka" if (os.getenv("CI", "false") == "true") else "127.0.0.1"
+    kafka_port = 9092 if (os.getenv("CI", "false") == "true") else 29092
+    log.info("Connecting to Kafka: %s:%s", kafka_host, kafka_port)
+
     consumer = KafkaConsumer(
         "dbserver1.inventory.customers",
-        bootstrap_servers=["kafka:9092"],
+        bootstrap_servers=[f"{kafka_host}:{kafka_port}"],
         auto_offset_reset="earliest",
         enable_auto_commit=False,
     )
@@ -112,7 +117,7 @@ def debezium(remote_pg: RemotePostgres):
     assert resp.status_code == 204
 
 
-def get_kafka_msg(consumer, ts_ms, before=None, after=None) -> None:
+def get_kafka_msg(consumer: KafkaConsumer, ts_ms, before=None, after=None) -> None:
     """
     Gets the message from Kafka and checks its validity
     Arguments:
@@ -124,6 +129,7 @@ def get_kafka_msg(consumer, ts_ms, before=None, after=None) -> None:
         after:    a dictionary, if not None, the after field from the kafka message must
                   have the same values for the same keys
     """
+    log.info("Bootstrap servers: %s", consumer.config["bootstrap_servers"])
     msg = consumer.poll()
     assert msg, "Empty message"
     for val in msg.values():