Merge pull request #47 from lalithkota/develop

OS Connector: Index Write method updated. Debezium Connector: Tombsto…
OpenG2P · Nov 2, 2024 · b88bea8 · b88bea8
2 parents b1d8dda + a1db072
commit b88bea8
Show file tree

Hide file tree

Showing 21 changed files with 81 additions and 52 deletions.
diff --git a/scripts/pbms/debezium-connectors/default.json b/scripts/pbms/debezium-connectors/default.json
@@ -14,7 +14,8 @@
         "topic.prefix": "${DB_PREFIX_INDEX}",
         "table.include.list": "public.res_partner,public.g2p_program,public.g2p_program_fund,public.g2p_program_membership,public.g2p_program_membership_duplicate,public.g2p_program_registrant_info,public.g2p_program_assessment,public.g2p_entitlement,public.g2p_payment_batch,public.g2p_payment",
         "heartbeat.interval.ms": "${DEFAULT_DEBEZIUM_CONNECTOR_HEARTBEAT_MS}",
-        "decimal.handling.mode": "double"
+        "decimal.handling.mode": "double",
+        "tombstones.on.delete": false
     },
     "wait_after_init_secs": 20
 }
diff --git a/scripts/pbms/opensearch-connectors/10.g2p_program.json b/scripts/pbms/opensearch-connectors/10.g2p_program.json
@@ -7,6 +7,7 @@
         "connection.password": "${OPENSEARCH_PASSWORD}",
         "tasks.max": "1",
         "topics": "${DB_PREFIX_INDEX}.public.g2p_program",
+        "index.write.method": "upsert",
         "key.ignore": "false",
         "schema.ignore": "true",
         "key.converter": "org.apache.kafka.connect.json.JsonConverter",
@@ -24,7 +25,7 @@
         "transforms.keyExtId.field": "id",
 
         "transforms.valExt.type": "org.openg2p.reporting.kafka.connect.transforms.ApplyJq${dollar}Value",
-        "transforms.valExt.expr": ".payload.after + {source_ts_ms: .payload.source.ts_ms}",
+        "transforms.valExt.expr": ".payload.after + (if .payload.after then {source_ts_ms: .payload.source.ts_ms} else null end)",
 
         "transforms.tsconvert01.type": "org.openg2p.reporting.kafka.connect.transforms.TimestampConverterAdv${dollar}Value",
         "transforms.tsconvert01.field": "source_ts_ms",

diff --git a/scripts/pbms/opensearch-connectors/20.res_partner.json b/scripts/pbms/opensearch-connectors/20.res_partner.json
@@ -7,6 +7,7 @@
         "connection.password": "${OPENSEARCH_PASSWORD}",
         "tasks.max": "1",
         "topics": "${DB_PREFIX_INDEX}.public.res_partner",
+        "index.write.method": "upsert",
         "key.ignore": "false",
         "schema.ignore": "true",
         "key.converter": "org.apache.kafka.connect.json.JsonConverter",
@@ -24,7 +25,7 @@
         "transforms.keyExtId.field": "id",
 
         "transforms.valExt.type": "org.openg2p.reporting.kafka.connect.transforms.ApplyJq${dollar}Value",
-        "transforms.valExt.expr": ".payload.after + {source_ts_ms: .payload.source.ts_ms}",
+        "transforms.valExt.expr": ".payload.after + (if .payload.after then {source_ts_ms: .payload.source.ts_ms} else null end)",
 
         "transforms.tsconvert01.type": "org.openg2p.reporting.kafka.connect.transforms.TimestampConverterAdv${dollar}Value",
         "transforms.tsconvert01.field": "source_ts_ms",

diff --git a/scripts/pbms/opensearch-connectors/21.res_partner_history.json b/scripts/pbms/opensearch-connectors/21.res_partner_history.json
@@ -24,7 +24,7 @@
         "transforms.keyExtId.field": "id",
 
         "transforms.valExt.type": "org.openg2p.reporting.kafka.connect.transforms.ApplyJq${dollar}Value",
-        "transforms.valExt.expr": ".payload.after + {source_ts_ms: .payload.source.ts_ms}",
+        "transforms.valExt.expr": ".payload.after + (if .payload.after then {source_ts_ms: .payload.source.ts_ms} else null end)",
 
         "transforms.tsconvert01.type": "org.openg2p.reporting.kafka.connect.transforms.TimestampConverterAdv${dollar}Value",
         "transforms.tsconvert01.field": "source_ts_ms",

diff --git a/scripts/pbms/opensearch-connectors/30.g2p_program_membership.json b/scripts/pbms/opensearch-connectors/30.g2p_program_membership.json
@@ -7,6 +7,7 @@
         "connection.password": "${OPENSEARCH_PASSWORD}",
         "tasks.max": "1",
         "topics": "${DB_PREFIX_INDEX}.public.g2p_program_membership",
+        "index.write.method": "upsert",
         "key.ignore": "false",
         "schema.ignore": "true",
         "key.converter": "org.apache.kafka.connect.json.JsonConverter",
@@ -24,7 +25,7 @@
         "transforms.keyExtId.field": "id",
 
         "transforms.valExt.type": "org.openg2p.reporting.kafka.connect.transforms.ApplyJq${dollar}Value",
-        "transforms.valExt.expr": ".payload.after + {source_ts_ms: .payload.source.ts_ms}",
+        "transforms.valExt.expr": ".payload.after + (if .payload.after then {source_ts_ms: .payload.source.ts_ms} else null end)",
 
         "transforms.tsconvert01.type": "org.openg2p.reporting.kafka.connect.transforms.TimestampConverterAdv${dollar}Value",
         "transforms.tsconvert01.field": "source_ts_ms",

diff --git a/scripts/pbms/opensearch-connectors/31.g2p_program_registrant_info.json b/scripts/pbms/opensearch-connectors/31.g2p_program_registrant_info.json
@@ -7,6 +7,7 @@
         "connection.password": "${OPENSEARCH_PASSWORD}",
         "tasks.max": "1",
         "topics": "${DB_PREFIX_INDEX}.public.g2p_program_registrant_info",
+        "index.write.method": "upsert",
         "key.ignore": "false",
         "schema.ignore": "true",
         "key.converter": "org.apache.kafka.connect.json.JsonConverter",
@@ -24,7 +25,7 @@
         "transforms.keyExtId.field": "id",
 
         "transforms.valExt.type": "org.openg2p.reporting.kafka.connect.transforms.ApplyJq${dollar}Value",
-        "transforms.valExt.expr": ".payload.after + {source_ts_ms: .payload.source.ts_ms}",
+        "transforms.valExt.expr": ".payload.after + (if .payload.after then {source_ts_ms: .payload.source.ts_ms} else null end)",
 
         "transforms.tsconvert01.type": "org.openg2p.reporting.kafka.connect.transforms.TimestampConverterAdv${dollar}Value",
         "transforms.tsconvert01.field": "source_ts_ms",

diff --git a/scripts/pbms/opensearch-connectors/32.g2p_program_registrant_info_history.json b/scripts/pbms/opensearch-connectors/32.g2p_program_registrant_info_history.json
@@ -24,7 +24,7 @@
         "transforms.keyExtId.field": "id",
 
         "transforms.valExt.type": "org.openg2p.reporting.kafka.connect.transforms.ApplyJq${dollar}Value",
-        "transforms.valExt.expr": ".payload.after + {source_ts_ms: .payload.source.ts_ms}",
+        "transforms.valExt.expr": ".payload.after + (if .payload.after then {source_ts_ms: .payload.source.ts_ms} else null end)",
 
         "transforms.tsconvert01.type": "org.openg2p.reporting.kafka.connect.transforms.TimestampConverterAdv${dollar}Value",
         "transforms.tsconvert01.field": "source_ts_ms",

diff --git a/scripts/pbms/opensearch-connectors/33.g2p_program_assessment.json b/scripts/pbms/opensearch-connectors/33.g2p_program_assessment.json
@@ -7,6 +7,7 @@
         "connection.password": "${OPENSEARCH_PASSWORD}",
         "tasks.max": "1",
         "topics": "${DB_PREFIX_INDEX}.public.g2p_program_assessment",
+        "index.write.method": "upsert",
         "key.ignore": "false",
         "schema.ignore": "true",
         "key.converter": "org.apache.kafka.connect.json.JsonConverter",
@@ -24,7 +25,7 @@
         "transforms.keyExtId.field": "id",
 
         "transforms.valExt.type": "org.openg2p.reporting.kafka.connect.transforms.ApplyJq${dollar}Value",
-        "transforms.valExt.expr": ".payload.after + {source_ts_ms: .payload.source.ts_ms}",
+        "transforms.valExt.expr": ".payload.after + (if .payload.after then {source_ts_ms: .payload.source.ts_ms} else null end)",
 
         "transforms.tsconvert01.type": "org.openg2p.reporting.kafka.connect.transforms.TimestampConverterAdv${dollar}Value",
         "transforms.tsconvert01.field": "source_ts_ms",

diff --git a/scripts/pbms/opensearch-connectors/40.g2p_program_membership_deduplicate.json b/scripts/pbms/opensearch-connectors/40.g2p_program_membership_deduplicate.json
@@ -7,6 +7,7 @@
         "connection.password": "${OPENSEARCH_PASSWORD}",
         "tasks.max": "1",
         "topics": "${DB_PREFIX_INDEX}.public.g2p_program_membership_duplicate",
+        "index.write.method": "upsert",
         "key.ignore": "false",
         "schema.ignore": "true",
         "key.converter": "org.apache.kafka.connect.json.JsonConverter",
@@ -24,7 +25,7 @@
         "transforms.keyExtId.field": "id",
 
         "transforms.valExt.type": "org.openg2p.reporting.kafka.connect.transforms.ApplyJq${dollar}Value",
-        "transforms.valExt.expr": ".payload.after + {source_ts_ms: .payload.source.ts_ms}",
+        "transforms.valExt.expr": ".payload.after + (if .payload.after then {source_ts_ms: .payload.source.ts_ms} else null end)",
 
         "transforms.tsconvert01.type": "org.openg2p.reporting.kafka.connect.transforms.TimestampConverterAdv${dollar}Value",
         "transforms.tsconvert01.field": "source_ts_ms",

diff --git a/scripts/pbms/opensearch-connectors/50.g2p_program_fund.json b/scripts/pbms/opensearch-connectors/50.g2p_program_fund.json
@@ -7,6 +7,7 @@
         "connection.password": "${OPENSEARCH_PASSWORD}",
         "tasks.max": "1",
         "topics": "${DB_PREFIX_INDEX}.public.g2p_program_fund",
+        "index.write.method": "upsert",
         "key.ignore": "false",
         "schema.ignore": "true",
         "key.converter": "org.apache.kafka.connect.json.JsonConverter",
@@ -24,7 +25,7 @@
         "transforms.keyExtId.field": "id",
 
         "transforms.valExt.type": "org.openg2p.reporting.kafka.connect.transforms.ApplyJq${dollar}Value",
-        "transforms.valExt.expr": ".payload.after + {source_ts_ms: .payload.source.ts_ms}",
+        "transforms.valExt.expr": ".payload.after + (if .payload.after then {source_ts_ms: .payload.source.ts_ms} else null end)",
 
         "transforms.tsconvert01.type": "org.openg2p.reporting.kafka.connect.transforms.TimestampConverterAdv${dollar}Value",
         "transforms.tsconvert01.field": "source_ts_ms",

diff --git a/scripts/pbms/opensearch-connectors/60.g2p_entitlement.json b/scripts/pbms/opensearch-connectors/60.g2p_entitlement.json
@@ -7,6 +7,7 @@
         "connection.password": "${OPENSEARCH_PASSWORD}",
         "tasks.max": "1",
         "topics": "${DB_PREFIX_INDEX}.public.g2p_entitlement",
+        "index.write.method": "upsert",
         "key.ignore": "false",
         "schema.ignore": "true",
         "key.converter": "org.apache.kafka.connect.json.JsonConverter",
@@ -24,7 +25,7 @@
         "transforms.keyExtId.field": "id",
 
         "transforms.valExt.type": "org.openg2p.reporting.kafka.connect.transforms.ApplyJq${dollar}Value",
-        "transforms.valExt.expr": ".payload.after + {source_ts_ms: .payload.source.ts_ms}",
+        "transforms.valExt.expr": ".payload.after + (if .payload.after then {source_ts_ms: .payload.source.ts_ms} else null end)",
 
         "transforms.tsconvert01.type": "org.openg2p.reporting.kafka.connect.transforms.TimestampConverterAdv${dollar}Value",
         "transforms.tsconvert01.field": "source_ts_ms",

diff --git a/scripts/pbms/opensearch-connectors/70.g2p_payment_batch.json b/scripts/pbms/opensearch-connectors/70.g2p_payment_batch.json
@@ -7,6 +7,7 @@
         "connection.password": "${OPENSEARCH_PASSWORD}",
         "tasks.max": "1",
         "topics": "${DB_PREFIX_INDEX}.public.g2p_payment_batch",
+        "index.write.method": "upsert",
         "key.ignore": "false",
         "schema.ignore": "true",
         "key.converter": "org.apache.kafka.connect.json.JsonConverter",
@@ -24,7 +25,7 @@
         "transforms.keyExtId.field": "id",
 
         "transforms.valExt.type": "org.openg2p.reporting.kafka.connect.transforms.ApplyJq${dollar}Value",
-        "transforms.valExt.expr": ".payload.after + {source_ts_ms: .payload.source.ts_ms}",
+        "transforms.valExt.expr": ".payload.after + (if .payload.after then {source_ts_ms: .payload.source.ts_ms} else null end)",
 
         "transforms.tsconvert01.type": "org.openg2p.reporting.kafka.connect.transforms.TimestampConverterAdv${dollar}Value",
         "transforms.tsconvert01.field": "source_ts_ms",

diff --git a/scripts/pbms/opensearch-connectors/80.g2p_payment.json b/scripts/pbms/opensearch-connectors/80.g2p_payment.json
@@ -7,6 +7,7 @@
         "connection.password": "${OPENSEARCH_PASSWORD}",
         "tasks.max": "1",
         "topics": "${DB_PREFIX_INDEX}.public.g2p_payment",
+        "index.write.method": "upsert",
         "key.ignore": "false",
         "schema.ignore": "true",
         "key.converter": "org.apache.kafka.connect.json.JsonConverter",
@@ -24,7 +25,7 @@
         "transforms.keyExtId.field": "id",
 
         "transforms.valExt.type": "org.openg2p.reporting.kafka.connect.transforms.ApplyJq${dollar}Value",
-        "transforms.valExt.expr": ".payload.after + {source_ts_ms: .payload.source.ts_ms}",
+        "transforms.valExt.expr": ".payload.after + (if .payload.after then {source_ts_ms: .payload.source.ts_ms} else null end)",
 
         "transforms.tsconvert01.type": "org.openg2p.reporting.kafka.connect.transforms.TimestampConverterAdv${dollar}Value",
         "transforms.tsconvert01.field": "source_ts_ms",

diff --git a/scripts/social-registry/debezium-connectors/default.json b/scripts/social-registry/debezium-connectors/default.json
@@ -12,9 +12,10 @@
         "database.password": "${DB_PASS}",
         "database.dbname": "${DB_NAME}",
         "topic.prefix": "${DB_PREFIX_INDEX}",
-        "table.include.list": "public.res_partner,public.g2p_id_type,public.g2p_reg_id",
+        "table.include.list": "public.g2p_region,public.g2p_id_type,public.res_partner,public.g2p_reg_id",
         "heartbeat.interval.ms": "${DEFAULT_DEBEZIUM_CONNECTOR_HEARTBEAT_MS}",
-        "decimal.handling.mode": "double"
+        "decimal.handling.mode": "double",
+        "tombstones.on.delete": false
     },
     "wait_after_init_secs": 20
 }
diff --git a/...opensearch-connectors/05.g2p_id_type.json → ...try/opensearch-connectors/05.id_type.json b/...opensearch-connectors/05.g2p_id_type.json → ...try/opensearch-connectors/05.id_type.json
@@ -1,12 +1,13 @@
 {
-    "name": "g2p_id_type_${DB_PREFIX_INDEX}",
+    "name": "id_type_${DB_PREFIX_INDEX}",
     "config": {
         "connector.class": "io.aiven.kafka.connect.opensearch.OpensearchSinkConnector",
         "connection.url": "${OPENSEARCH_URL}",
         "connection.username": "${OPENSEARCH_USERNAME}",
         "connection.password": "${OPENSEARCH_PASSWORD}",
         "tasks.max": "1",
         "topics": "${DB_PREFIX_INDEX}.public.g2p_id_type",
+        "index.write.method": "upsert",
         "key.ignore": "false",
         "schema.ignore": "true",
         "key.converter": "org.apache.kafka.connect.json.JsonConverter",
@@ -24,7 +25,7 @@
         "transforms.keyExtId.field": "id",
 
         "transforms.valExt.type": "org.openg2p.reporting.kafka.connect.transforms.ApplyJq${dollar}Value",
-        "transforms.valExt.expr": ".payload.after + {source_ts_ms: .payload.source.ts_ms}",
+        "transforms.valExt.expr": ".payload.after + (if .payload.after then {source_ts_ms: .payload.source.ts_ms} else null end)",
 
         "transforms.tsconvert01.type": "org.openg2p.reporting.kafka.connect.transforms.TimestampConverterAdv${dollar}Value",
         "transforms.tsconvert01.field": "source_ts_ms",
@@ -41,5 +42,5 @@
         "transforms.tsSelect.ts.order": "write_date,create_date,source_ts_ms",
         "transforms.tsSelect.output.field": "@timestamp_gen"
     },
-    "wait_after_init_secs": 15
+    "wait_after_init_secs": 20
 }
diff --git a/...ch-connectors/06.g2p_id_type_history.json → ...search-connectors/06.id_type_history.json b/...ch-connectors/06.g2p_id_type_history.json → ...search-connectors/06.id_type_history.json
@@ -1,5 +1,5 @@
 {
-    "name": "g2p_id_type_history_${DB_PREFIX_INDEX}",
+    "name": "id_type_history_${DB_PREFIX_INDEX}",
     "config": {
         "connector.class": "io.aiven.kafka.connect.opensearch.OpensearchSinkConnector",
         "connection.url": "${OPENSEARCH_URL}",
@@ -24,7 +24,7 @@
         "transforms.keyExtId.field": "id",
 
         "transforms.valExt.type": "org.openg2p.reporting.kafka.connect.transforms.ApplyJq${dollar}Value",
-        "transforms.valExt.expr": ".payload.after + {source_ts_ms: .payload.source.ts_ms}",
+        "transforms.valExt.expr": ".payload.after + (if .payload.after then {source_ts_ms: .payload.source.ts_ms} else null end)",
 
         "transforms.tsconvert01.type": "org.openg2p.reporting.kafka.connect.transforms.TimestampConverterAdv${dollar}Value",
         "transforms.tsconvert01.field": "source_ts_ms",

diff --git a/...opensearch-connectors/10.res_partner.json → ...opensearch-connectors/11.res_partner.json b/...opensearch-connectors/10.res_partner.json → ...opensearch-connectors/11.res_partner.json
@@ -7,6 +7,7 @@
         "connection.password": "${OPENSEARCH_PASSWORD}",
         "tasks.max": "1",
         "topics": "${DB_PREFIX_INDEX}.public.res_partner",
+        "index.write.method": "upsert",
         "key.ignore": "false",
         "schema.ignore": "true",
         "key.converter": "org.apache.kafka.connect.json.JsonConverter",
@@ -24,7 +25,7 @@
         "transforms.keyExtId.field": "id",
 
         "transforms.valExt.type": "org.openg2p.reporting.kafka.connect.transforms.ApplyJq${dollar}Value",
-        "transforms.valExt.expr": ".payload.after + {source_ts_ms: .payload.source.ts_ms}",
+        "transforms.valExt.expr": ".payload.after + (if .payload.after then {source_ts_ms: .payload.source.ts_ms} else null end)",
 
         "transforms.tsconvert01.type": "org.openg2p.reporting.kafka.connect.transforms.TimestampConverterAdv${dollar}Value",
         "transforms.tsconvert01.field": "source_ts_ms",
@@ -50,5 +51,6 @@
         "transforms.tsSelect.type": "org.openg2p.reporting.kafka.connect.transforms.TimestampSelector${dollar}Value",
         "transforms.tsSelect.ts.order": "write_date,create_date,source_ts_ms",
         "transforms.tsSelect.output.field": "@timestamp_gen"
-    }
+    },
+    "wait_after_init_secs": 60
 }
diff --git a/...ch-connectors/11.res_partner_history.json → ...ch-connectors/12.res_partner_history.json b/...ch-connectors/11.res_partner_history.json → ...ch-connectors/12.res_partner_history.json
@@ -24,7 +24,7 @@
         "transforms.keyExtId.field": "id",
 
         "transforms.valExt.type": "org.openg2p.reporting.kafka.connect.transforms.ApplyJq${dollar}Value",
-        "transforms.valExt.expr": ".payload.after + {source_ts_ms: .payload.source.ts_ms}",
+        "transforms.valExt.expr": ".payload.after + (if .payload.after then {source_ts_ms: .payload.source.ts_ms} else null end)",
 
         "transforms.tsconvert01.type": "org.openg2p.reporting.kafka.connect.transforms.TimestampConverterAdv${dollar}Value",
         "transforms.tsconvert01.field": "source_ts_ms",

diff --git a/...rch-connectors/12.res_partner_dedupe.json → ...rch-connectors/13.res_partner_dedupe.json b/...rch-connectors/12.res_partner_dedupe.json → ...rch-connectors/13.res_partner_dedupe.json
@@ -27,7 +27,7 @@
         "transforms.keyExtId.field": "id",
 
         "transforms.valExt.type": "org.openg2p.reporting.kafka.connect.transforms.ApplyJq${dollar}Value",
-        "transforms.valExt.expr": ".payload.after + {source_ts_ms: .payload.source.ts_ms}",
+        "transforms.valExt.expr": ".payload.after + (if .payload.after then {source_ts_ms: .payload.source.ts_ms} else null end)",
 
         "transforms.renameTopic.type": "org.openg2p.reporting.kafka.connect.transforms.RenameTopic",
         "transforms.renameTopic.topic": "res_partner"