Documents are not deleted from the index

[SergofSergs]

Hello, we are currently trying to implement CDC with pgsync between postgres-15 and opensearch.
INSERTs and UPDATEs are working fine, no problem with them, but DELETE operation are not synced at all.

Am I missing some configuration?

Here is my docker compose file:

services:
  redis:
    image: redis:8.0.3-alpine
    container_name: redis
    ports:
      - 6379:6379
    volumes:
      - redis-data:/data

  postgres:
    image: postgres:15
    container_name: postgres
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: postgres
    ports:
      - 5433:5432
    command: ["postgres", "-c", "wal_level=logical", "-c", "max_replication_slots=1", "-c", "max_slot_wal_keep_size=10240"]

  opensearch:
    image: opensearchproject/opensearch:latest
    container_name: opensearch
    environment:
      discovery.type: single-node
      node.name: opensearch
      bootstrap.memory_lock: true
      OPENSEARCH_JAVA_OPTS: "-Xms512m -Xmx512m"
      OPENSEARCH_INITIAL_ADMIN_PASSWORD: "CBsZ6#9kHxb"
      DISABLE_SECURITY_PLUGIN: true
    ulimits:
      memlock:
        soft: -1
        hard: -1
      nofile:
        soft: 65536 # maximum number of open files for the OpenSearch user, set to at least 65536 on modern systems
        hard: 65536
    volumes:
      - opensearch-data:/usr/share/opensearch/data
    ports:
      - 9200:9200
      - 9600:9600
    healthcheck:
      test: curl -s -f http://localhost:9200/_cluster/health?wait_for_status=yellow >/dev/null || exit 1
      interval: 5s
      retries: 3
      start_period: 30s

  opensearch-dashboards:
    image: opensearchproject/opensearch-dashboards:latest
    container_name: opensearch-dashboards
    ports:
      - 5601:5601
    expose:
      - "5601"
    environment:
      OPENSEARCH_HOSTS: '["http://opensearch:9200"]'
      DISABLE_SECURITY_DASHBOARDS_PLUGIN: true
    depends_on:
      - opensearch

  pg-sync:
    image: toluaina1/pgsync:latest
    container_name: pg-sync
    environment:
      SCHEMA: /schema/schema.json
      PG_HOST: postgres
      PG_PORT: 5432
      PG_USER: postgres
      PG_PASSWORD: postgres
      OPENSEARCH: true
      ELASTICSEARCH: false
      ELASTICSEARCH_SCHEME: http
      ELASTICSEARCH_HOST: opensearch
      ELASTICSEARCH_PORT: 9200
      ELASTICSEARCH_USER: admin
      ELASTICSEARCH_PASSWORD: "CBsZ6#9kHxb"
      ELASTICSEARCH_VERIFY_CERTS: false
      REDIS_CHECKPOINT: true
      REDIS_HOST: redis
      REDIS_PORT: 6379
    volumes:
      - ./schema.json:/schema/schema.json
    depends_on:
      redis:
        condition: service_started
      postgres:
        condition: service_started
      opensearch:
        condition: service_healthy
    command: ["-b", "-d"]

volumes:
  redis-data:
  opensearch-data:

This is part of wal of a table content, here you can see that DELETE operations are present

My schema.json

[
  {
    "database": "postgres",
    "index": "shipment",
    "nodes": {
      "table": "shipment",
      "schema": "oms_shipments",
      "columns": [
        "id",
        "order_number",
        "shipment_number",
        "merchant_id",
        "client_id",
        "provider_id",
        "status",
        "built_date_time",
        "tracking_number",
        "additional_tracking_number",
        "service_code",
        "delivery_type_id",
        "first_mile_type",
        "point_code",
        "drop_off_code",
        "req_delivery_date",
        "delivery_plan_date",
        "shipment_weight",
        "shipment_length",
        "shipment_width",
        "shipment_height",
        "shipment_volume_weight",
        "shipment_declare_currency",
        "sender",
        "receiver",
        "products",
        "shipment_date_creation",
        "box_quantity",
        "provider_delivery_price_rub",
        "qcg",
        "not_for_personal_use",
        "mutable"
      ],
      "transform": {
        "mapping": {
          "id": {"type": "long"},
          "order_number": {"type": "keyword"},
          "shipment_number": {"type": "keyword"},
          "merchant_id": {"type": "keyword"},
          "client_id": {"type": "keyword"},
          "provider_id": {"type": "keyword"},
          "status": {"type": "keyword"},
          "built_date_time": {"type": "date"},
          "tracking_number": {"type": "keyword"},
          "additional_tracking_number": {"type": "keyword"},
          "service_code": {"type": "keyword"},
          "delivery_type_id": {"type": "keyword"},
          "first_mile_type": {"type": "keyword"},
          "point_code": {"type": "keyword"},
          "drop_off_code": {"type": "keyword"},
          "req_delivery_date": {"type": "date"},
          "delivery_plan_date": {"type": "date"},
          "shipment_weight": {"type": "long"},
          "shipment_length": {"type": "long"},
          "shipment_width": {"type": "long"},
          "shipment_height": {"type": "long"},
          "shipment_volume_weight": {"type": "double"},
          "shipment_declare_currency": {"type": "keyword"},
          "sender": {
            "type": "nested",
            "properties": {
              "placeId": {"type": "keyword"}
            }
          },
          "receiver": {"type": "nested"},
          "products": {"type": "nested"},
          "shipment_date_creation": {"type": "date"},
          "box_quantity": {"type": "long"},
          "provider_delivery_price_rub": {"type": "double"},
          "qcg": {"type": "boolean"},
          "not_for_personal_use": {"type": "boolean"},
          "mutable": {"type": "boolean"}
        }
      }
    }
  }
]

[toluaina]

The -b option was added recently to pgsync as a convenience. It allows you to bootstrap and run pgsync in one go.

However, it can be like a footgun if you re-run pgsync -b -d all repeatedly: each time you do, it tears down and recreates the replication slot.

That is the most likely cause of the behavior you're seeing. You stop the container, delete some records in the DB, and then restart, but the WAL entries for those deletions are no longer available.

It may be better if the behavior were to bootstrap only when necessary. (note taken)

For now, can you ensure you use the -b option only once, and that subsequent restarts omit it?

[SergofSergs]

Thanks for the reply!
So it seems that problem got away when I upped pgsync version to 7.0.1, before i used some 2 month old by now version.

And i also did what you recommended also, now my container tracking if is it bootstrap or some subsequent run.

New modifying Dokerfile:

FROM toluaina1/pgsync:latest

USER root

RUN mkdir -p /app/data/init
RUN chown -R pgsync:pgsync /app/data/init && chmod 700 /app/data/init

VOLUME /app/data/init

USER pgsync

My new pg-sync service:

pg-sync:
  build:
    context: .
    dockerfile: Dockerfile.pgsync
  container_name: pg-sync
  environment:
    SCHEMA: /schema/schema.json
    PG_HOST: postgres
    PG_PORT: 5432
    PG_USER: postgres
    PG_PASSWORD: postgres
    OPENSEARCH: true
    ELASTICSEARCH: false
    ELASTICSEARCH_SCHEME: http
    ELASTICSEARCH_HOST: opensearch
    ELASTICSEARCH_PORT: 9200
    ELASTICSEARCH_USER: admin
    ELASTICSEARCH_PASSWORD: "CBsZ6#9kHxb"
    ELASTICSEARCH_VERIFY_CERTS: false
    REDIS_CHECKPOINT: true
    REDIS_HOST: redis
    REDIS_PORT: 6379
  volumes:
    - ./schema.json:/schema/schema.json
  depends_on:
    redis:
      condition: service_started
    postgres:
      condition: service_started
    opensearch:
      condition: service_healthy
  entrypoint: /bin/sh
  command: >
      -c "
            if [ ! -f /app/data/init/.init ]; then
              echo 'Bootstrap run';
              if pgsync -b; then
                  echo 'Bootstrap completed';
                  touch /app/data/init/.init;
                  exec pgsync -d;
              else
                  echo 'Bootstrap failed. Exiting...';
                  exit 1;
              fi
            else
              echo 'Subsequent run';
              exec pgsync -d;
            fi
      "

P.S. Do not close the issue now, because I have not tested this on my production stand :D. If it's gone for good I'll close this in the nearest future