Filter out change events of logical decoding messages in user.source.streaming connector

Author: rkudryashov

compose.yaml

@@ -122,11 +122,13 @@ services:
       OFFSET_STORAGE_TOPIC: kafka-connect.offset
       STATUS_STORAGE_TOPIC: kafka-connect.status
       ENABLE_APICURIO_CONVERTERS: true
+      ENABLE_DEBEZIUM_SCRIPTING: true
       CONNECT_EXACTLY_ONCE_SOURCE_SUPPORT: enabled
       CONNECT_CONFIG_PROVIDERS: "file"
       CONNECT_CONFIG_PROVIDERS_FILE_CLASS: "org.apache.kafka.common.config.provider.FileConfigProvider"
       CONNECT_LOG4J_LOGGER_org.apache.kafka.clients: ERROR
     volumes:
+      - ./kafka-connect/filtering/groovy:/kafka/connect/debezium-connector-postgres/filtering/groovy
       - ./kafka-connect/postgres.properties:/secrets/postgres.properties:ro
       - ./kafka-connect/logs/:/kafka/logs/
 

kafka-connect/connectors/user.source.streaming.json

@@ -18,7 +18,10 @@
     "column.exclude.list": "public.library_user.created_at,public.library_user.updated_at",
     "heartbeat.interval.ms": "5000",
     "publication.autocreate.mode": "filtered",
-    "transforms": "renameTopic",
+    "transforms": "filter,renameTopic",
+    "transforms.filter.type": "io.debezium.transforms.Filter",
+    "transforms.filter.language": "jsr223.groovy",
+    "transforms.filter.condition": "valueSchema.field('op') != null && value.op != 'm'",
     "transforms.renameTopic.type": "io.debezium.transforms.ByLogicalTableRouter",
     "transforms.renameTopic.topic.regex": "(.*)public.(.*)",
     "transforms.renameTopic.topic.replacement": "$1users",