POST http://localhost:8083/connectors
{
    "name": "source-connector",
    "config": {
        "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
        "plugin.name": "pgoutput",
        "database.hostname": "chatting-db-2",
        "database.port": "5434",
        "database.user": "postgres",
        "database.password": "password",
        "database.dbname" : "chat2",
        "database.server.name": "dbserver5434",
        "transforms": "unwrap,addTopicPrefix",

        # message의 schema를 after 필드로만 설정
        # 이 부분을 설정하지 않는다면 source와 sink connector의 schema가 일치하지 않는다.
        # 즉, kafka로 메세지가 흘러갈 순 있지만 kafka에서 db로 sink가 진행되지 않을것이다.
        "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
        "transforms.addTopicPrefix.type":"org.apache.kafka.connect.transforms.RegexRouter",
        "transforms.addTopicPrefix.regex":"(.*)",
        "transforms.addTopicPrefix.replacement":"$1"
    }
}

[
    {
        "class": "io.confluent.connect.jdbc.JdbcSinkConnector",
        "type": "sink",
        "version": "10.6.0"
    },
    {
        "class": "io.confluent.connect.jdbc.JdbcSourceConnector",
        "type": "source",
        "version": "10.6.0"
    },
    ...
    {
        "class": "io.debezium.connector.postgresql.PostgresConnector",
        "type": "source",
        "version": "1.9.7.Final"
    },
    ...
]

POST http://localhost:8083/connectors
{
    "name": "sink-connector",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
        "task.max" : 1,
        "topics": "dbserver5434.public.user_table",

        "connection.url": "jdbc:postgresql://chatting-db-1:5433/chat1",
        "connection.user":"postgres",
        "connection.password":"password",

        # table/column 자동생성 방지
        # 두개의 테이블이 이미 동일함
        "auto.create": "false",
        "auto.evolve": "false",
        "delete.enabled": "true",
        "insert.mode": "upsert",
        "pk.mode": "record_key",
        "tombstones.on.delete": "true",

        # schema일치 확인 및 payload 추출 과정
        "key.converter": "org.apache.kafka.connect.json.JsonConverter",
        "key.converter.schemas.enable": "true",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "value.converter.schemas.enable": "true",
        "transforms": "unwrap,addTopicPrefix",
        "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
        "transforms.addTopicPrefix.type":"org.apache.kafka.connect.transforms.RegexRouter",
        "transforms.addTopicPrefix.regex":"(.*)",
        "transforms.addTopicPrefix.replacement":"$1",

        # 해당 테이블에 new row 삽입
        "table.name.format":"user_table",

        # 몇 개의 메세지를 읽고 sink할 것인지
        "batch.size": "1"
    }
}

{
   "schema": {
      "type": "struct",
      "fields": [
         {
            "type": "string",
            "optional": false,
            "field": "user_id"
         },
         {
            "type": "string",
            "optional": true,
            "field": "user_name"
         },
         {
            "type": "string",
            "optional": true,
            "field": "user_status"
         }
      ],
      "optional": false,
      "name": "dbserver5434.public.user_table.Value"
   },
   "payload": {
      "user_id": "a",
      "user_name": "a",
      "user_status": "a"
   }
}

gyuminhwangbo@Gyuminui-MacBookPro spring-chatting-server % sh ./install-jdbc-connector.sh

(step-1) confluent-hub cli 다운로드 및 압축해제
  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100 7584k  100 7584k    0     0  1131k      0  0:00:06  0:00:06 --:--:-- 1587k
mkdir: confluent-etcs: File exists
x share/doc/confluent-hub-client/notices/
x share/doc/confluent-hub-client/licenses/
x share/java/confluent-hub-client/jakarta.ws.rs-api-2.1.6.jar
...

(step-2) confluent-hub 환경변수 설정
/Users/gyuminhwangbo/study/spring-chatting-server

(step-3) cli를 통한 jdbc connector 다운로드
Running in a "--no-prompt" mode
Implicit acceptance of the license below:
Confluent Community License
https://www.confluent.io/confluent-community-license
Downloading component Kafka Connect JDBC 10.6.0, provided by Confluent, Inc. from Confluent Hub and installing into /Users/gyuminhwangbo/study/spring-chatting-server/confluent-etcs
Implicit confirmation of the question: Do you want to uninstall existing version 10.6.0?
...

(step-4) debezium connector 컨테이너의 connector리스트에 삽입

(step-5) debezium connector 컨테이너 재시작으로 loading new connector
postgres-kafka-source-connector

# (1) kafka-connector가 postgresDB:5434의 Txlog에서 변경사항 감지(CDC)
postgres-kafka-source-connector | 2023-01-08 07:19:07,589 INFO   Postgres|dbserver5434|streaming  First LSN 'LSN{0/168BF58}' received   [io.debezium.connector.postgresql.connection.WalPositionLocator]
postgres-kafka-source-connector | 2023-01-08 07:19:07,589 INFO   Postgres|dbserver5434|streaming  WAL resume position 'LSN{0/168BF58}' discovered   [io.debezium.connector.postgresql.PostgresStreamingChangeEventSource]
postgres-kafka-source-connector | 2023-01-08 07:19:07,593 INFO   Postgres|dbserver5434|streaming  Connection gracefully closed   [io.debezium.jdbc.JdbcConnection]
postgres-kafka-source-connector | 2023-01-08 07:19:07,600 INFO   Postgres|dbserver5434|streaming  Connection gracefully closed   [io.debezium.jdbc.JdbcConnection]

# (Additional) POST 반환값
nginx                     | 192.168.240.1 - - [08/Jan/2023:07:19:07 +0000] "POST /chat/user HTTP/1.1" 200 86 "-" "PostmanRuntime/7.29.2" "-"

# (2) PgOutput - postgres가 기본적으로 제공하는 replica설정을 앞서 우리가
# logical로 바꿧었다. 그리고 logical로 저장된 TXlog들을 디코딩해서 kafka에 밀어넣기 위해
# PgOutput이라는 모듈을 통해 logicalTX----(decoding-PgOutput)---->kafka를 수행한다.
postgres-kafka-source-connector | 2023-01-08 07:19:07,614 INFO   Postgres|dbserver5434|streaming  Initializing PgOutput logical decoder publication   [io.debezium.connector.postgresql.connection.PostgresReplicationConnection]
chatting-db-2             | 2023-01-08 07:19:07.620 UTC [82] LOG:  starting logical decoding for slot "debezium"
chatting-db-2             | 2023-01-08 07:19:07.620 UTC [82] DETAIL:  Streaming transactions committing after 0/168BDD8, reading WAL from 0/168BDD8.
chatting-db-2             | 2023-01-08 07:19:07.620 UTC [82] STATEMENT:  START_REPLICATION SLOT "debezium" LOGICAL 0/168BDD8 ("proto_version" '1', "publication_names" 'dbz_publication')
chatting-db-2             | 2023-01-08 07:19:07.620 UTC [82] LOG:  logical decoding found consistent point at 0/168BDD8
chatting-db-2             | 2023-01-08 07:19:07.620 UTC [82] DETAIL:  There are no running transactions.
chatting-db-2             | 2023-01-08 07:19:07.620 UTC [82] STATEMENT:  START_REPLICATION SLOT "debezium" LOGICAL 0/168BDD8 ("proto_version" '1', "publication_names" 'dbz_publication')
postgres-kafka-source-connector | 2023-01-08 07:19:07,636 INFO   Postgres|dbserver5434|streaming  Requested thread factory for connector PostgresConnector, id = dbserver5434 named = keep-alive   [io.debezium.util.Threads]
postgres-kafka-source-connector | 2023-01-08 07:19:07,637 INFO   Postgres|dbserver5434|streaming  Creating thread debezium-postgresconnector-dbserver5434-keep-alive   [io.debezium.util.Threads]
postgres-kafka-source-connector | 2023-01-08 07:19:07,638 INFO   Postgres|dbserver5434|streaming  Processing messages   [io.debezium.connector.postgresql.PostgresStreamingChangeEventSource]
postgres-kafka-source-connector | 2023-01-08 07:19:07,654 INFO   Postgres|dbserver5434|streaming  Message with LSN 'LSN{0/168BF58}' arrived, switching off the filtering   [io.debezium.connector.postgresql.connection.WalPositionLocator]
postgres-kafka-source-connector | 2023-01-08 07:19:08,264 INFO   ||  1 records sent during previous 00:01:18.579, last recorded offset: {transaction_id=null, lsn_proc=23641944, lsn=23641944, txId=501, ts_usec=1673162347279067}   [io.debezium.connector.common.BaseSourceTask]

# (Additional) 이건 그냥 별개로 connector 안거치고 바로 kafka에 삽입하는 별도의 pipeline.
chatting-server-2         | 2023-01-08 07:19:07.806  INFO 1 --- [ad | producer-1] chatting.chat.web.ChatController         : 메세지 전송 성공 topic=log-user-add, offset=0, partition=1

# (3) JDBC-Sink connector가 kafka-topic의 소비된 message의 last offset을 확인하고,
# 신규 데이터 발견, postgresql에 대한 dialect를 만들어서 쿼리를 실행시키는 과정
postgres-kafka-source-connector | 2023-01-08 07:19:08,287 INFO   ||  [Producer clientId=connector-producer-source-connector-0] Resetting the last seen epoch of partition dbserver5434.public.user_table-0 to 0 since the associated topicId changed from null to oCJqYdENQ1C2cPZeNEhtsw   [org.apache.kafka.clients.Metadata]
postgres-kafka-source-connector | 2023-01-08 07:19:08,313 INFO   ||  Attempting to open connection #1 to PostgreSql   [io.confluent.connect.jdbc.util.CachedConnectionProvider]
postgres-kafka-source-connector | 2023-01-08 07:19:08,412 INFO   ||  Maximum table name length for database is 63 bytes   [io.confluent.connect.jdbc.dialect.PostgreSqlDatabaseDialect]
postgres-kafka-source-connector | 2023-01-08 07:19:08,412 INFO   ||  JdbcDbWriter Connected   [io.confluent.connect.jdbc.sink.JdbcDbWriter]
postgres-kafka-source-connector | 2023-01-08 07:19:08,430 INFO   ||  Checking PostgreSql dialect for existence of TABLE "user_table"   [io.confluent.connect.jdbc.dialect.GenericDatabaseDialect]
postgres-kafka-source-connector | 2023-01-08 07:19:08,440 INFO   ||  Using PostgreSql dialect TABLE "user_table" present   [io.confluent.connect.jdbc.dialect.GenericDatabaseDialect]
postgres-kafka-source-connector | 2023-01-08 07:19:08,456 INFO   ||  Checking PostgreSql dialect for type of TABLE "user_table"   [io.confluent.connect.jdbc.dialect.GenericDatabaseDialect]
postgres-kafka-source-connector | 2023-01-08 07:19:08,460 INFO   ||  Setting metadata for table "user_table" to Table{name='"user_table"', type=TABLE columns=[Column{'user_name', isPrimaryKey=false, allowsNull=true, sqlType=varchar}, Column{'user_id', isPrimaryKey=true, allowsNull=false, sqlType=varchar}, Column{'user_status', isPrimaryKey=false, allowsNull=true, sqlType=varchar}]}   [io.confluent.connect.jdbc.util.TableDefinitions]