update all

Debezium/LICENSE

@@ -0,0 +1,3 @@
+Additional permission under GNU GPL version 3 section 7
+
+If you modify this Program, or any covered work, by linking or combining it with [name of library] (or a modified version of that library), containing parts covered by the terms of [name of library's license], the licensors of this Program grant you additional permission to convey the resulting work. Corresponding Source for a non-source form of such a combination shall include the source code for the parts of [name of library] used as well as that of the covered work.

Debezium/README.md

@@ -0,0 +1,38 @@
+## Debezium Postgres CDC
+
+This module provides a Debezium connector for Postgres. It is based on the [Debezium Postgres connector](https://debezium.io/documentation/reference/1.0/connectors/postgresql.html)
+and uses confluent zookeeper, broker, schema registry and rest proxy.
+
+### Prerequisites
+- Docker
+- Docker Compose
+
+
+### Running the Stack
+- Run `docker-compose up` to start the connector
+- Run `docker-compose down` to stop the connector
+- Run `docker-compose logs -f` to view the logs
+- Run `docker-compose exec postgres bash` to access the postgres container
+- Run `docker-compose exec broker bash` to access the kafka container
+
+
+### Registering the connector
+- Run `curl -X POST -H "Content-Type: application/json" -d @connector.json http://localhost:8083/connectors`
+
+### Testing the connector
+- Run `docker-compose exec postgres bash` to access the postgres container
+- Run `psql -U postgres` to access the postgres database
+
+### Viewing the data
+- Run `docker-compose exec broker bash` to access the kafka container
+- Run `kafka-console-consumer --bootstrap-server broker:9092 --topic postgres.public.movies --from-beginning`
+
+
+## Schema Registry
+
+### Viewing the schema
+- Run `curl -X GET http://localhost:8081/subjects/postgres.public.movies/versions/1`
+
+
+### Registering the schema
+- Run `curl -X POST -H "Content-Type: application/json" -d @schema.json http://localhost:8081/subjects/postgres.public.movies/versions`

Debezium/connector.json

@@ -0,0 +1,23 @@
+{
+  "name": "movies-db-connector",
+  "config": {
+    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
+    "plugin.name": "pgoutput",
+    "tasks.max": "1",
+    "database.hostname": "postgres",
+    "database.port": "5432",
+    "database.user": "postgres",
+    "database.password": "postgres",
+    "database.dbname": "movies_db",
+    "database.server.name": "postgres",
+    "table.include.list": "public.movies",
+    "database.history.kafka.bootstrap.servers": "kafka:9092",
+    "database.history.kafka.topic": "schema-changes.movies",
+    "topic.prefix": "postgres",
+    "topic.creation.enable": "true",
+    "topic.creation.default.replication.factor": "1",
+    "topic.creation.default.partitions": "1",
+    "topic.creation.default.cleanup.policy": "delete",
+    "topic.creation.default.retention.ms": "604800000"
+  }
+}

Debezium/docker-compose.yml

@@ -0,0 +1,141 @@
+version: "3.9"
+
+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
+    healthcheck:
+      test: echo srvr | nc zookeeper 2181 || exit 1
+      start_period: 10s
+      retries: 20
+      interval: 10s
+  broker:
+    image: confluentinc/cp-kafka:7.3.1
+    hostname: broker
+    container_name: broker
+    depends_on:
+      zookeeper:
+        condition: service_healthy
+    ports:
+      - "29092:29092"
+      - "9092:9092"
+      - "9101:9101"
+    environment:
+      KAFKA_BROKER_ID: 1
+      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost: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: localhost
+    healthcheck:
+      test: nc -z localhost 9092 || exit -1
+      start_period: 15s
+      interval: 5s
+      timeout: 10s
+      retries: 10
+  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:29092
+      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: 10s
+      interval: 10s
+      timeout: 5s
+      retries: 5
+
+  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:29092
+      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
+
+  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:29092'
+      KAFKA_REST_LISTENERS: "http://0.0.0.0:8082"
+
+  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
+
+
+  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

Debezium/scripts/data.sql

@@ -0,0 +1,15 @@
+
+CREATE TABLE IF NOT EXISTS movies (
+    id bigint PRIMARY KEY,
+    title VARCHAR(255) NOT NULL,
+    year INTEGER NOT NULL,
+    director VARCHAR(255) NOT NULL,
+    rating DECIMAL NOT NULL
+);
+
+INSERT INTO movies (id, title, year, director, rating) VALUES (1, 'The Shawshank Redemption', 1994, 'Frank Darabont', 9.3);
+INSERT INTO movies (id, title, year, director, rating) VALUES (2, 'The Godfather', 1972, 'Francis Ford Coppola', 9.2);
+INSERT INTO movies (id, title, year, director, rating) VALUES (3, 'The Godfather: Part II', 1974, 'Francis Ford Coppola', 9.0);
+INSERT INTO movies (id, title, year, director, rating) VALUES (4, 'The Dark Knight', 2008, 'Christopher Nolan', 9.0);
+INSERT INTO movies (id, title, year, director, rating) VALUES (5, 'Interstellar', 2014, 'Christopher Nolan', 10);
+