title: Postgres Debezium CDC does not publish changes to Kafka
body: <p>My current testing configuration looks like so:</p>
  
  <pre><code>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
  </code></pre>
  
  <p>I run it with docker-compose like so:</p>
  
  <pre><code>$ docker-compose up
  </code></pre>
  
  <p>And I see no error messages. It seems like everything is running ok. If I do <code>docker ps</code>, I see that all services are running. </p>
  
  <p>In order to check that Kafka is running, I made Kafka producer and Kafka consumer in Python:</p>
  
  <pre><code># 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)
  </code></pre>
  
  <p>And it works absolutely great. I see how my producer publishes messages and I see how they are consumed in consumer window. </p>
  
  <p>Now I want to make CDC work. First of all, inside Postgres container I set <code>postgres</code> role password to <code>postgres</code>:</p>
  
  <pre><code>$ su postgres
  $ psql
  psql&gt; \password postgres
  Enter new password: postgres
  </code></pre>
  
  <p>I then created a new database <code>test</code>:</p>
  
  <pre><code>psql&gt; CREATE DATABASE test;
  </code></pre>
  
  <p>I created a table:</p>
  
  <pre><code>psql&gt; \c test;
  test=# create table mytable (id serial, name varchar(128), primary key(id));
  </code></pre>
  
  <p>And, finally, for my Debezium CDC stack I created a connector:</p>
  
  <pre><code>$ 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"}
  </code></pre>
  
  <p>As you can see, my connector was created without any errors. Now I expect Debezium CDC to publish all changes to Kafka topic <code>public.some_topic</code>. To check this, I create a new Kafka comsumer:</p>
  
  <pre><code>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)
  </code></pre>
  
  <p>The only difference with the first example, is that I'm watching <code>public.some_topic</code>. I then go to database console and make an insert:</p>
  
  <pre><code>test=# insert into mytable (name) values ('Tom Cat');    
  INSERT 0 1
  test=#
  </code></pre>
  
  <p>So, a new value is inserted, but I see nothing is happening in consumer window. In other words, Debezium does not publish events to Kafka <code>public.some_topic</code>. What is wrong with that and how can I fix it?</p>