onkarantad/kafka-cdc repository overview

docker compose -f postgres-cdc-compose.yaml up -d

docker exec -it kafka-cdc bash

cd /opt/bitnami/cdc

kafka-topics.sh --create --topic dbhistory.history001 --bootstrap-server localhost:9092

nohup connect-standalone.sh /opt/bitnami/cdc/worker.properties /opt/bitnami/cdc/src.properties /opt/bitnami/cdc/sink.properties &> /bitnami/cdc/logs/cdc1.log &

docker stop kafka-cdc postgres-src postgres-tgt && docker rm -f kafka-cdc postgres-src postgres-tgt && docker volume rm postgres-cdc_kafka-data postgres-cdc_postgres-data-src postgres-cdc_postgres-data-tgt

services:
  postgres-src:
    image: onkarantad/postgres:0.0.2
    container_name: postgres-src
    restart: always
    ports:
      - "5432:5432"
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
    command:
      - "postgres"
      - "-c"
      - "wal_level=logical"
    volumes:
      - postgres-data-src:/var/lib/postgresql/data
    networks:
      - local-network
      
  postgres-tgt:
    image: onkarantad/postgres:0.0.2
    container_name: postgres-tgt
    restart: always
    depends_on:
      - postgres-src
    ports:
      - "5433:5432"
    environment:
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
    command:
      - "postgres"
      - "-c"
      - "wal_level=logical"
    volumes:
      - postgres-data-tgt:/var/lib/postgresql/data
    networks:
      - local-network
      
  kafka-cdc:
    image: onkarantad/kafka-cdc:0.0.2
    container_name: kafka-cdc
    restart: always
    #hostname: localhost
    depends_on:
      - postgres-src
      - postgres-tgt
    user: root
    environment:
      # KAFKA
      - HOSTNAME=kafka-cdc
      - PORT=9092
      # SRC DATABASE
      - DB_SRC_HOST=postgres-src
      - DB_SRC_PORT=5432
      - DB_SRC=postgres
      - DB_SRC_USER=postgres
      - DB_SRC_PASS=postgres
      # TGT DATABASE 
      - DB_TGT_HOST=postgres-tgt
      - DB_TGT_PORT=5432
      - DB_TGT=postgres
      - DB_TGT_USER=postgres
      - DB_TGT_PASS=postgres
      # SRC PROPERTIES
      - TABLE_WHITELIST=employee,dept
      - DB_HISTORY_KAFKA_TOPIC=dbhistory.history001
      - TOPIC_PREFIX=public
      # SINK PROPERTIES
      - TOPICS=employee
      - topic=$${topic}
      # KRaft settings
      - KAFKA_CFG_NODE_ID=0
      - KAFKA_CFG_PROCESS_ROLES=controller,broker
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=0@kafka-cdc:9093
      # Listeners
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://:9092
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=PLAINTEXT
    command: >
      /bin/bash -c "envsubst < /opt/bitnami/cdc/worker1.properties > /opt/bitnami/cdc/worker.properties &&
                    envsubst < /opt/bitnami/cdc/src1.properties > /opt/bitnami/cdc/src.properties &&
                    envsubst < /opt/bitnami/cdc/sink1.properties > /opt/bitnami/cdc/sink.properties &&
                    exec /opt/bitnami/scripts/kafka/run.sh"
    networks:
      - local-network
    volumes:
      - "kafka-data:/bitnami"
    ports:
      - "9092:9092"
      
volumes:
  postgres-data-src:
  postgres-data-tgt:
  kafka-data:
    driver: local
networks:
  local-network:
    external: true

create table employee(
id int primary key,
name varchar(255),
salary int,
update_date timestamp
);

insert into employee values (101,'aa',500000,'2012-01-01 01:01:01');
insert into employee values (102,'bb',500000,'2012-01-01 01:01:01');
insert into employee values (103,'cc',500000,'2012-01-01 01:01:01');
insert into employee values (104,'dd',500000,'2012-01-01 01:01:01');
insert into employee values (105,'dd',500000,'2012-01-01 01:01:01');
insert into employee values (106,'dd',500000,'2012-01-01 01:01:01');
insert into employee values (107,'dd',500000,'2012-01-01 01:01:01');
insert into employee values (108,'dd',500000,'2012-01-01 01:01:01');
insert into employee values (109,'dd',500000,'2012-01-01 01:01:01');
insert into employee values (110,'dd',500000,'2012-01-01 01:01:01');