• docker-compose configuration.

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.3.1
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - '2181:2181'
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
      ALLOW_ANONYMOUS_LOGIN: yes  
    healthcheck:
      test: nc -z localhost 2181 || exit -1
      interval: 10s
      retries: 3
      start_period: 10s
      timeout: 5s
    networks:
      - kafka-network

  broker:
    image: confluentinc/cp-kafka:7.3.1
    hostname: broker
    container_name: broker
    depends_on:
      zookeeper:
        condition: service_healthy
    ports:
      - '9092:9092'
      - '9101:9101'
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://<server-ip>:9092
      KAFKA_LISTENERS: PLAINTEXT://0.0.0.0:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'true'
      KAFKA_JMX_PORT: 9101
      KAFKA_JMX_HOSTNAME: <server-ip>
      KAFKA_CONFLUENT_SUPPORT_METRICS_ENABLE: "false"
      KAFKA_MESSAGE_MAX_BYTES: 2000000000
      KAFKA_REPLICA_FETCH_MAX_BYTES: 2000000000
      KAFKA_REPLICA_MAX_LOG_SIZE: 2000000000  
    healthcheck:
      test: ["CMD-SHELL", "nc -z localhost 9092"]
      interval: 10s
      retries: 3
      start_period: 30s
      timeout: 5s
    networks:
      - kafka-network

  debezium:
    image: debezium/connect:latest
    restart: always
    container_name: debezium
    hostname: debezium
    depends_on:
      postgres:
        condition: service_healthy
      broker:
        condition: service_healthy
    ports:
      - '8083:8083'
    environment:
      BOOTSTRAP_SERVERS: broker:9092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: connect_configs
      STATUS_STORAGE_TOPIC: connect_statuses
      OFFSET_STORAGE_TOPIC: connect_offsets
      KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      ENABLE_DEBEZIUM_SCRIPTING: 'true'
    healthcheck:
      test:
        [
          'CMD',
          'curl',
          '--silent',
          '--fail',
          '-X',
          'GET',
          '<http://localhost:8083/connectors>',
        ]
      start_period: 30s
      interval: 30s
      timeout: 10s
      retries: 10
    networks:
      - kafka-network

  schema-registry:
    image: confluentinc/cp-schema-registry:7.3.1
    hostname: schema-registry
    container_name: schema-registry
    depends_on:
      broker:
        condition: service_healthy
    ports:
      - '8081:8081'
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: broker:9092
      SCHEMA_REGISTRY_LISTENERS: <http://0.0.0.0:8081>
    healthcheck:
      start_period: 10s
      interval: 10s
      retries: 20
      test: curl --user superUser:superUser --fail --silent --insecure <http://localhost:8081/subjects> --output /dev/null || exit 1
    networks:
      - kafka-network

  rest-proxy:
    image: confluentinc/cp-kafka-rest:7.3.1
    depends_on:
      broker:
        condition: service_healthy
    ports:
      - '8082:8082'
    hostname: rest-proxy
    container_name: rest-proxy
    environment:
      KAFKA_REST_HOST_NAME: rest-proxy
      KAFKA_REST_BOOTSTRAP_SERVERS: 'broker:9092'
      KAFKA_REST_LISTENERS: '<http://0.0.0.0:8082>'
    networks:
      - kafka-network

  debezium-ui:
    image: debezium/debezium-ui:latest
    restart: always
    container_name: debezium-ui
    hostname: debezium-ui
    depends_on:
      debezium:
        condition: service_healthy
    ports:
      - '8080:8080'
    environment:
      KAFKA_CONNECT_URIS: <http://debezium:8083>
    networks:
      - kafka-network

  postgres:
    image: postgres:latest
    restart: always
    container_name: postgres
    hostname: postgres
    ports:
      - '5432:5432'
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: movies_db
    command: ['postgres', '-c', 'wal_level=logical']
    healthcheck:
      test: ['CMD', 'psql', '-U', 'postgres', '-c', 'SELECT 1']
      interval: 10s
      timeout: 5s
      retries: 5
    volumes:
      - ./scripts:/docker-entrypoint-initdb.d
    networks:
      - kafka-network

networks:
  kafka-network:
    driver: bridge

volumes:
  postgres-data:
    driver: local

• Connector Configuration: postgres-connector-config.json

{
  "name": "postgres-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",
    "database.hostname": "[host_name]",
    "database.port": "[port]",
    "database.user": "[db_user]",
    "database.password": "[db_pwd]",
    "database.dbname": "[db_name]",
    "database.server.name": "[postgres_server_name]",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_slot",
    "publication.name": "debezium_pub",
    "table.include.list": "[schema].[table_name]",
    "database.history.kafka.bootstrap.servers": "broker:9092",
    "database.history.kafka.topic": "schema-changes.[table_name]",
    "topic.prefix": "[topic]"
  }
}

• Post create connector

curl -X POST -H "Content-Type: application/json" --data @postgres-connector-config.json <http://localhost:8083/connectors>

Or

curl -X POST <http://localhost:8083/connectors> -H "Content-Type: application/json" -d '{
  "name": "postgres-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",
    "database.hostname": "[host_name]",
    "database.port": "[port]",
    "database.user": "[db_user]",
    "database.password": "[db_pwd]",
    "database.dbname": "[db_name]",
    "database.server.name": "[postgres_server_name]",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_slot",
    "publication.name": "debezium_pub",
    "table.include.list": "[schema].[table_name]",
    "database.history.kafka.bootstrap.servers": "broker:9092",
    "database.history.kafka.topic": "schema-changes.[table_name]",
    "topic.prefix": "[topic]"
  }
}'