Postgres Дебезиум CD C не публикуется sh меняется на Кафку - PullRequest
Новая
       10

Postgres Дебезиум CD C не публикуется sh меняется на Кафку

1 голос
/ 15 января 2020

Моя текущая конфигурация тестирования выглядит так: 

version: '3.7'
services:
  postgres:
    image: debezium/postgres
    restart: always
    ports:
      - "5432:5432"
  zookeeper:
    image: debezium/zookeeper
    ports:
      - "2181:2181"
      - "2888:2888"
      - "3888:3888"
  kafka:
    image: debezium/kafka
    restart: always
    ports:
      - "9092:9092"
    links:
      - zookeeper
    depends_on:
      - zookeeper
    environment:
     - ZOOKEEPER_CONNECT=zookeeper:2181
     - KAFKA_GROUP_MIN_SESSION_TIMEOUT_MS=250
  connect:
    image: debezium/connect
    restart: always
    ports:
      - "8083:8083"
    links:
      - zookeeper
      - postgres
      - kafka
    depends_on:
      - zookeeper
      - postgres
      - kafka
    environment:
      - BOOTSTRAP_SERVERS=kafka:9092
      - GROUP_ID=1
      - CONFIG_STORAGE_TOPIC=my_connect_configs
      - OFFSET_STORAGE_TOPIC=my_connect_offsets
      - STATUS_STORAGE_TOPIC=my_source_connect_statuses

Я запускаю ее с docker -составить так: 

$ docker-compose up

И я не вижу сообщений об ошибках. Кажется, все работает нормально. Если я делаю docker ps, я вижу, что все службы работают.

Чтобы проверить, работает ли Kafka, я сделал производителя Kafka и потребителя Kafka в Python: 

# producer. I run it in one console window
from kafka import KafkaProducer
from json import dumps
from time import sleep

producer = KafkaProducer(bootstrap_servers=['localhost:9092'], value_serializer=lambda x: dumps(x).encode('utf-8'))

for e in range(1000):
    data = {'number' : e}
    producer.send('numtest', value=data)
    sleep(5)

# consumer. I run it in other colsole window

from kafka import KafkaConsumer
from json import loads

consumer = KafkaConsumer(
    'numtest',
     bootstrap_servers=['localhost:9092'],
     auto_offset_reset='earliest',
     enable_auto_commit=True,
     group_id='my-group',
     value_deserializer=lambda x: loads(x.decode('utf-8')))

for message in consumer:
     print(message)

И он работает абсолютно великолепно. Я вижу, как мой производитель публикует сообщения, и вижу, как они потребляются в окне потребителя.

Теперь я хочу, чтобы CD C работал. Прежде всего, внутри Postgres контейнера я установил postgres пароль роли на postgres: 

$ su postgres
$ psql
psql> \password postgres
Enter new password: postgres

Затем я создал новую базу данных test: 

psql> CREATE DATABASE test;

I создал таблицу: 

psql> \c test;
test=# create table mytable (id serial, name varchar(128), primary key(id));

И, наконец, для моего стека Debezium CD C я создал разъем: 

$ curl -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '{
    "name": "test-connector",
    "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",
    "plugin.name": "pgoutput",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "postgres",
    "database.password": "postgres",
    "database.dbname" : "test",
    "database.server.name": "postgres",
    "database.whitelist": "public.mytable",
    "database.history.kafka.bootstrap.servers": "localhost:9092",
    "database.history.kafka.topic": "public.some_topic"
    }
}'

{"name":"test-connector","config":{"connector.class":"io.debezium.connector.postgresql.PostgresConnector","tasks.max":"1","plugin.name":"pgoutput","database.hostname":"postgres","database.port":"5432","database.user":"postgres","database.password":"postgres","database.dbname":"test","database.server.name":"postgres","database.whitelist":"public.mytable","database.history.kafka.bootstrap.servers":"localhost:9092","database.history.kafka.topic":"public.some_topic","name":"test-connector"},"tasks":[],"type":"source"}

Как видите, мой разъем был создан без каких-либо ошибки. Теперь я ожидаю, что Debezium CD C опубликует sh все изменения в Kafka topi c public.some_topic. Чтобы проверить это, я создаю новый потребитель Kafka: 

from kafka import KafkaConsumer
from json import loads

consumer = KafkaConsumer(
    'public.some_topic',
     bootstrap_servers=['localhost:9092'],
     auto_offset_reset='earliest',
     enable_auto_commit=True,
     group_id='my-group',
     value_deserializer=lambda x: loads(x.decode('utf-8')))

for message in consumer:
     print(message)

Единственное отличие от первого примера в том, что я смотрю public.some_topic. Затем я go в консоль базы данных и вставляю: 

test=# insert into mytable (name) values ('Tom Cat');    
INSERT 0 1
test=#

Итак, новое значение вставлено, но я вижу, что в окне потребителя ничего не происходит. Другими словами, Дебезиум не публикует события sh для Кафки public.some_topic. Что не так с этим и как я могу это исправить?

1 Ответ

4 голосов
/ 15 января 2020

Использование вашего Docker Compose Я вижу эту ошибку в рабочем журнале Kafka Connect при создании соединителя: 

Caused by: org.postgresql.util.PSQLException: ERROR: could not access file "pgoutput": No such file or directory
        at org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2505)
        at org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2241)
        at org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:310)
        at org.postgresql.jdbc.PgStatement.executeInternal(PgStatement.java:447)
        at org.postgresql.jdbc.PgStatement.execute(PgStatement.java:368)
        at org.postgresql.jdbc.PgStatement.executeWithFlags(PgStatement.java:309)
        at org.postgresql.jdbc.PgStatement.executeCachedSql(PgStatement.java:295)
        at org.postgresql.jdbc.PgStatement.executeWithFlags(PgStatement.java:272)
        at org.postgresql.jdbc.PgStatement.execute(PgStatement.java:267)
        at io.debezium.connector.postgresql.connection.PostgresReplicationConnection.createReplicationSlot(PostgresReplicationConnection.java:288)
        at io.debezium.connector.postgresql.PostgresConnectorTask.start(PostgresConnectorTask.java:126)
        ... 9 more

Это также отражается в статусе задачи, если вы используете Kafka Connect REST API для запроса: 

curl -s "http://localhost:8083/connectors?expand=info&expand=status" | jq '."test-connector".status'
{
  "name": "test-connector",
  "connector": {
    "state": "RUNNING",
    "worker_id": "192.168.16.5:8083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "FAILED",
      "worker_id": "192.168.16.5:8083",
      "trace": "org.apache.kafka.connect.errors.ConnectException: org.postgresql.util.PSQLException: ERROR: could not access file \"pgoutput\": No such file or directory\n\tat io.debezium.connector.postgresql.PostgresConnectorTask.start(PostgresConnectorTask.java:129)\n\tat io.debezium.connector.common.BaseSourceTask.start(BaseSourceTask.java:49)\n\tat org.apache.kafka.connect.runtime.WorkerSourceTask.execute(WorkerSourceTask.java:208)\n\tat org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:177)\n\tat org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:227)\n\tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)\n\tat java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)\n\tat java.base/java.lang.Thread.run(Thread.java:834)\nCaused by: org.postgresql.util.PSQLException: ERROR: could not access file \"pgoutput\": No such file or directory\n\tat org.postgresql.core.v3.QueryExecutorImpl.receiveErrorResponse(QueryExecutorImpl.java:2505)\n\tat org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2241)\n\tat org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:310)\n\tat org.postgresql.jdbc.PgStatement.executeInternal(PgStatement.java:447)\n\tat org.postgresql.jdbc.PgStatement.execute(PgStatement.java:368)\n\tat org.postgresql.jdbc.PgStatement.executeWithFlags(PgStatement.java:309)\n\tat org.postgresql.jdbc.PgStatement.executeCachedSql(PgStatement.java:295)\n\tat org.postgresql.jdbc.PgStatement.executeWithFlags(PgStatement.java:272)\n\tat org.postgresql.jdbc.PgStatement.execute(PgStatement.java:267)\n\tat io.debezium.connector.postgresql.connection.PostgresReplicationConnection.createReplicationSlot(PostgresReplicationConnection.java:288)\n\tat io.debezium.connector.postgresql.PostgresConnectorTask.start(PostgresConnectorTask.java:126)\n\t... 9 more\n"
    }
  ],
  "type": "source"

Используемая версия Postgres: 

postgres=# SHOW server_version;
 server_version
----------------
 9.6.16

pgoutput доступен только> = версия 10.

Я изменил ваш Docker Compose для использования версии 10: 

image: debezium/postgres:10

После отскока стека для чистого запуска и следуя вашим инструкциям, я получаю соединитель, который работает: 

curl -s "http://localhost:8083/connectors?expand=info&expand=status" | \
           jq '. | to_entries[] | [ .value.info.type, .key, .value.status.connector.state,.value.status.tasks[].state,.value.info.config."connector.class"]|join(":|:")' | \
           column -s : -t| sed 's/\"//g'| sort
source  |  test-connector  |  RUNNING  |  RUNNING  |  io.debezium.connector.postgresql.PostgresConnector

и данные в топике Кафки c: 

$ docker exec kafkacat kafkacat -b kafka:9092 -t postgres.public.mytable -C
{"schema":{"type":"struct","fields":[{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":true,"field":"name"}],"optional":true,"name":"postgres.public.mytable.Value","field":"before"},{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"},{"type":"string","optional":true,"field":"name"}],"optional":true,"name":"postgres.public.mytable.Value","field":"after"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":false,"field":"schema"},{"type":"string","optional":false,"field":"table"},{"type":"int64","optional":true,"field":"txId"},{"type":"int64","optional":true,"field":"lsn"},{"type":"int64","optional":true,"field":"xmin"}],"optional":false,"name":"io.debezium.connector.postgresql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"}],"optional":false,"name":"postgres.public.mytable.Envelope"},"payload":{"before":null,"after":{"id":1,"name":"Tom Cat"},"source":{"version":"1.0.0.Final","connector":"postgresql","name":"postgres","ts_ms":1579172192292,"snapshot":"false","db":"test","schema":"public","table":"mytable","txId":561,"lsn":24485520,"xmin":null},"op":"c","ts_ms":1579172192347}}% Reached end of topic postgres.public.mytable [0] at offset 1

Я добавил kafkacat в ваш Docker Составить с: 

  kafkacat:
    image: edenhill/kafkacat:1.5.0
    container_name: kafkacat
    entrypoint: 
      - /bin/sh 
      - -c 
      - |
        while [ 1 -eq 1 ];do sleep 60;done

Редактировать : сохранение предыдущего ответа, поскольку оно все еще полезно и актуально:

Дебезиум запишет сообщение в topi c на основе имени таблицы . В вашем примере это будет postgres.test.mytable.

Вот почему kafkacat полезен, потому что вы можете запустить 

kafkacat -b broker:9092 -L

, чтобы увидеть список всех ваших тем и разделов. Как только у вас есть topi c

kafkacat -b broker:9092 -t postgres.test.mytable -C

для чтения из него.

Проверьте подробности по kafkacat , включая инструкции по запуску с Docker

Также есть демонстрационная версия все в действии с Docker Составьте здесь

...