اتصال RDB ها و موتورهای جستجو – فصل 3

docker compose up -d

-- Create Debezium user
CREATE ROLE debezium WITH LOGIN PASSWORD 'dbz' REPLICATION;

-- Create target table
CREATE TABLE testtable (
  id INTEGER PRIMARY KEY,
  message VARCHAR(255)
);
INSERT INTO testtable (id, message) VALUES (1, 'CDC test row');

-- Grant SELECT permission
GRANT SELECT ON testtable TO debezium;

-- Create publication (for pgoutput plugin)
CREATE PUBLICATION debezium_pub FOR TABLE testtable;

docker compose down --volumes

command: >
  postgres
  -c wal_level=logical
  -c max_replication_slots=4
  -c max_wal_senders=4

docker compose restart postgres

docker compose down --volumes

curl -X POST "localhost:8083/connectors" \
  -H "Content-Type: application/json" \
  -d '{
    "name": "postgres-connector",
    "config": {
      "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
      "database.hostname": "postgres",
      "database.port": "5432",
      "database.user": "debezium",
      "database.password": "dbz",
      "database.dbname": "postgres",
      "database.server.name": "dbserver1",
      "topic.prefix": "dbserver1",
      "plugin.name": "pgoutput",
      "publication.name": "debezium_pub",
      "slot.name": "debezium_slot",
      "slot.drop.on.stop": "true",
      "table.include.list": "public.testtable",
      "snapshot.mode": "initial",
      "tombstones.on.delete": "false",
      "key.converter": "org.apache.kafka.connect.json.JsonConverter",
      "value.converter": "org.apache.kafka.connect.json.JsonConverter",
      "key.converter.schemas.enable": "false",
      "value.converter.schemas.enable": "false"
    }
  }'

curl http://localhost:8083/connectors/postgres-connector/status

{
  "name": "postgres-connector",
  "connector": {
    "state": "RUNNING",
    "worker_id": "10.4.1.81:8083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "RUNNING",
      "worker_id": "10.4.1.81:8083"
    }
  ],
  "type": "source"
}

# Enter Kafka container
docker compose exec kafka bash

# List topics
kafka-topics --bootstrap-server localhost:9092 --list

dbserver1.public.testtable

kafka-console-consumer --bootstrap-server localhost:9092 \
  --topic dbserver1.public.testtable \
  --from-beginning

{
  "before": null,
  "after": {
    "id": 1,
    "message": "CDC test row"
  },
  "op": "r"
}

# Log into PostgreSQL
docker compose exec postgres psql -U postgres

# Insert new row
INSERT INTO testtable (id, message) VALUES (2, 'inserted row');

{
  "before": null,
  "after": {
    "id": 2,
    "message": "inserted row"
  },
  "op": "c"
}

{
  "before": null,
  "after": {
    "id": 2,
    "message": "inserted row"
  },
  "op": "c"
}

SELECT * FROM pg_publication;

docker compose down --volumes
docker compose up -d

docker compose logs connect

kafka-console-consumer --bootstrap-server localhost:9092 \
  --topic dbserver1.public.testtable \
  --from-beginning