Enforce exactly-once upsert writes using table primary key (#45)

* Enforce upsert semantics using pk

* Bump release version

bin/test

@@ -12,15 +12,13 @@ function debug {
     make kubectl get pod -- -A | grep -i metabase-app-setup | awk '{ print $2 }' | xargs -I{} make kubectl logs pod/{} -- -n dataplane
 }
 
-trap debug EXIT
-
 make template &&
     make molecule destroy &&
     make molecule create &&
     make images &&
     make molecule converge &&
-    make wait &&
-    make molecule verify &&
-    make molecule side-effect &&
-    make molecule verify &&
-    trap - EXIT
+    make wait
+
+trap debug EXIT
+make molecule verify
+trap - EXIT

charts/dataplane/Chart.yaml

@@ -16,10 +16,10 @@ type: application
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
 # Versions are expected to follow Semantic Versioning (https://semver.org/)
-version: 0.1.16
+version: 0.1.17
 
 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application. Versions are not expected to
 # follow Semantic Versioning. They should reflect the version the application is using.
 # It is recommended to use it with quotes.
-appVersion: 0.1.16
+appVersion: 0.1.17

charts/dataplane/templates/strimzi/connector.yaml

@@ -61,6 +61,7 @@ spec:
     database.dbname: "{{ $db.dbname }}"
     database.server.name: "{{ $dbId }}"
     snapshot.new.tables: parallel
+    snapshot.mode: {{ $db.snapshot_mode | default "initial" }}
     slot.name: "{{ $dbSlotName }}"
     plugin.name: pgoutput
     tasks.max: "1"
@@ -109,6 +110,7 @@ spec:
     database.dbname: "{{ $db.dbname }}"
     database.server.name: "{{ $dbId }}-{{ $partition.sink }}"
     snapshot.new.tables: parallel
+    snapshot.mode: {{ $db.snapshot_mode | default "initial" }}
     slot.name: {{ $partitionSlotName }}
     plugin.name: pgoutput
     tasks.max: "1"
@@ -147,6 +149,8 @@ spec:
     tasks.max: "1"
     topics.regex: "cdc.(.*)"
     auto.create: true
+    pk.mode: "record_key"
+    pk.fields: ""
     key.converter: io.confluent.connect.avro.AvroConverter
     key.converter.schema.registry.url: http://{{ $registryName }}:{{ $registryPort }}
     value.converter: io.confluent.connect.avro.AvroConverter

charts/dataplane/values.yaml

@@ -7,7 +7,7 @@ util:
   image:
     repository: nephelaiio/dataplane-util
     pullPolicy: IfNotPresent
-    tag: dataplane-0.1.16
+    tag: dataplane-0.1.17
   resources: {}
 
 cdc:
@@ -45,7 +45,7 @@ strimzi:
   connect:
     image:
       repository: nephelaiio/dataplane-connect
-      tag: dataplane-0.1.16
+      tag: dataplane-0.1.17
     replicas: 1
     config:
       group.id: connect-cluster
@@ -100,7 +100,7 @@ metabase:
     image:
       repository: nephelaiio/dataplane-util
       pullPolicy: IfNotPresent
-      tag: dataplane-0.1.16
+      tag: dataplane-0.1.17
     securityContext: {}
 
   image:

molecule/default/molecule.yml

@@ -66,6 +66,7 @@ provisioner:
           dataplane_chart: dataplane
           dataplane_path: ${MOLECULE_PROJECT_DIRECTORY}
           dataplane_metabase_hostname: "metabase.{{ k8s_cluster_name }}"
+          dataplane_pagila_connector_name: pagila-connector
           dataplane_chart_values:
             util:
               image:
@@ -91,7 +92,7 @@ provisioner:
                 - hostname: "{{ dataplane_pagila_team }}-{{ dataplane_pagila_db }}"
                   port: 5432
                   id: pagila
-                  connector: pagila-connector
+                  connector: "{{ dataplane_pagila_connector_name }}"
                   dbname: "{{ dataplane_pagila_db }}"
                   signaling: "{{ dataplane_pagila_signaling }}"
                   exclude:

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "dataplane"
-version = "0.1.16"
+version = "0.1.17"
 description = ""
 authors = ["Teodoro Cook <[email protected]>"]