본문 바로가기

Open Source/Kafka

Postgres Source Connector

{
"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"transforms.TimestampConverter.target.type": "Timestamp",
"connection.password": "******",
"transforms.TimestampConverter.field": "update_date",
"tasks.max": "1",
"transforms": "unwrap,route,TimestampConverter",
"mode": "bulk",
"tombstones.on.delete": "true",
"transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
"schema.registry.url": "http://schema-registry:8081",
"transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
"auto.evolve": "false",
"transforms.unwrap.drop.tombstones": "true",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"value.converter": "io.confluent.connect.avro.AvroConverter",
"insert.mode": "upsert",
"transforms.route.replacement": "$3",
"key.converter": "io.confluent.connect.avro.AvroConverter",
"table.name.format": "${topic}",
"transforms.TimestampConverter.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
"transforms.TimestampConverter.format": "yyyy-MM-ddHH:mm:ss",
"topics.regex": "postgres.(.*)",
"value.converter.schema.registry.url": "http://schema-registry:8081",
"key.converter.schemas.enable": "false",
"delete.enabled": "true",
"connection.user": "yajcemulnsbppm",
"value.converter.schemas.enable": "true",
"name": "PostgreSourceConnector3",
"auto.create": "true",
"connection.url": "jdbc:postgresql://ec2-3-224-125-117.compute-1.amazonaws.com:5432/d6360me1rf3uoq",
"key.converter.schema.registry.url": "http://schema-registry:8081",
"pk.mode": "record_key"
}

 

 

query mode : https://docs.confluent.io/kafka-connectors/jdbc/current/source-connector/source_config_options.html  Mode로 검색