Skip to content

Commit 7bc5336

Browse files
teoincontattoteoincontatto
committed
feat: support for skipping drop of replication slot and publication on tombstone
1 parent 69726e4 commit 7bc5336

8 files changed

Lines changed: 78 additions & 8 deletions

File tree

‎stackgres-k8s/src/common/src/main/java/io/stackgres/common/crd/sgstream/StackGresStreamSourcePostgres.java‎

Lines changed: 17 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,8 @@ public class StackGresStreamSourcePostgres {
4242

4343
private List<String> excludes;
4444

45+
private Boolean skipDropReplicationSlotAndPublicationOnTombstone;
46+
4547
@Valid
4648
private StackGresStreamSourcePostgresDebeziumProperties debeziumProperties;
4749

@@ -101,6 +103,15 @@ public void setExcludes(List<String> excludes) {
101103
this.excludes = excludes;
102104
}
103105

106+
public Boolean getSkipDropReplicationSlotAndPublicationOnTombstone() {
107+
return skipDropReplicationSlotAndPublicationOnTombstone;
108+
}
109+
110+
public void setSkipDropReplicationSlotAndPublicationOnTombstone(
111+
Boolean skipDropReplicationSlotAndPublicationOnTombstone) {
112+
this.skipDropReplicationSlotAndPublicationOnTombstone = skipDropReplicationSlotAndPublicationOnTombstone;
113+
}
114+
104115
public StackGresStreamSourcePostgresDebeziumProperties getDebeziumProperties() {
105116
return debeziumProperties;
106117
}
@@ -112,8 +123,8 @@ public void setDebeziumProperties(
112123

113124
@Override
114125
public int hashCode() {
115-
return Objects.hash(database, debeziumProperties, excludes, host, includes, password,
116-
port, username);
126+
return Objects.hash(database, debeziumProperties, excludes, host, includes, password, port,
127+
skipDropReplicationSlotAndPublicationOnTombstone, username);
117128
}
118129

119130
@Override
@@ -129,7 +140,10 @@ public boolean equals(Object obj) {
129140
&& Objects.equals(debeziumProperties, other.debeziumProperties)
130141
&& Objects.equals(excludes, other.excludes) && Objects.equals(host, other.host)
131142
&& Objects.equals(includes, other.includes) && Objects.equals(password, other.password)
132-
&& Objects.equals(port, other.port) && Objects.equals(username, other.username);
143+
&& Objects.equals(port, other.port)
144+
&& Objects.equals(skipDropReplicationSlotAndPublicationOnTombstone,
145+
other.skipDropReplicationSlotAndPublicationOnTombstone)
146+
&& Objects.equals(username, other.username);
133147
}
134148

135149
@Override

‎stackgres-k8s/src/common/src/main/java/io/stackgres/common/crd/sgstream/StackGresStreamSourceSgCluster.java‎

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,8 @@ public class StackGresStreamSourceSgCluster {
4040

4141
private List<String> excludes;
4242

43+
private Boolean skipDropReplicationSlotAndPublicationOnTombstone;
44+
4345
@Valid
4446
private StackGresStreamSourcePostgresDebeziumProperties debeziumProperties;
4547

@@ -91,6 +93,15 @@ public void setExcludes(List<String> excludes) {
9193
this.excludes = excludes;
9294
}
9395

96+
public Boolean getSkipDropReplicationSlotAndPublicationOnTombstone() {
97+
return skipDropReplicationSlotAndPublicationOnTombstone;
98+
}
99+
100+
public void setSkipDropReplicationSlotAndPublicationOnTombstone(
101+
Boolean skipDropReplicationSlotAndPublicationOnTombstone) {
102+
this.skipDropReplicationSlotAndPublicationOnTombstone = skipDropReplicationSlotAndPublicationOnTombstone;
103+
}
104+
94105
public StackGresStreamSourcePostgresDebeziumProperties getDebeziumProperties() {
95106
return debeziumProperties;
96107
}
@@ -103,7 +114,7 @@ public void setDebeziumProperties(
103114
@Override
104115
public int hashCode() {
105116
return Objects.hash(database, debeziumProperties, excludes, includes, name, password,
106-
username);
117+
skipDropReplicationSlotAndPublicationOnTombstone, username);
107118
}
108119

109120
@Override
@@ -119,6 +130,8 @@ public boolean equals(Object obj) {
119130
&& Objects.equals(debeziumProperties, other.debeziumProperties)
120131
&& Objects.equals(excludes, other.excludes) && Objects.equals(includes, other.includes)
121132
&& Objects.equals(name, other.name) && Objects.equals(password, other.password)
133+
&& Objects.equals(skipDropReplicationSlotAndPublicationOnTombstone,
134+
other.skipDropReplicationSlotAndPublicationOnTombstone)
122135
&& Objects.equals(username, other.username);
123136
}
124137

‎stackgres-k8s/src/common/src/main/resources/crds/SGStream.yaml‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -140,6 +140,9 @@ spec:
140140
type: string
141141
description: |
142142
A regular expressions that allow to match one or more `<schema>.<table>.<column>` entries to be filtered out before sending to the target.
143+
skipDropReplicationSlotAndPublicationOnTombstone:
144+
type: boolean
145+
description: When set to `true` replication slot and publication will not be dropped after receiving the tombstone signal.
143146
debeziumProperties: &source-postgres-debeziumProperties
144147
type: object
145148
description: |

stackgres-k8s/src/operator/src/main/java/io/stackgres/operator/validation/stream/LockValidator.java renamed to stackgres-k8s/src/operator/src/main/java/io/stackgres/operator/validation/stream/StatusLockValidator.java

File renamed without changes.

‎stackgres-k8s/src/restapi/src/main/java/io/stackgres/apiweb/dto/stream/StreamSourcePostgres.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,8 @@ public class StreamSourcePostgres {
3030

3131
private List<String> excludes;
3232

33+
private Boolean skipDropReplicationSlotAndPublicationOnTombstone;
34+
3335
private StreamSourcePostgresDebeziumProperties debeziumProperties;
3436

3537
private String usernameValue;
@@ -92,6 +94,15 @@ public void setExcludes(List<String> excludes) {
9294
this.excludes = excludes;
9395
}
9496

97+
public Boolean getSkipDropReplicationSlotAndPublicationOnTombstone() {
98+
return skipDropReplicationSlotAndPublicationOnTombstone;
99+
}
100+
101+
public void setSkipDropReplicationSlotAndPublicationOnTombstone(
102+
Boolean skipDropReplicationSlotAndPublicationOnTombstone) {
103+
this.skipDropReplicationSlotAndPublicationOnTombstone = skipDropReplicationSlotAndPublicationOnTombstone;
104+
}
105+
95106
public StreamSourcePostgresDebeziumProperties getDebeziumProperties() {
96107
return debeziumProperties;
97108
}

‎stackgres-k8s/src/restapi/src/main/java/io/stackgres/apiweb/dto/stream/StreamSourceSgCluster.java‎

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,8 @@ public class StreamSourceSgCluster {
2828

2929
private List<String> excludes;
3030

31+
private Boolean skipDropReplicationSlotAndPublicationOnTombstone;
32+
3133
private StreamSourcePostgresDebeziumProperties debeziumProperties;
3234

3335
private String usernameValue;
@@ -82,6 +84,15 @@ public void setExcludes(List<String> excludes) {
8284
this.excludes = excludes;
8385
}
8486

87+
public Boolean getSkipDropReplicationSlotAndPublicationOnTombstone() {
88+
return skipDropReplicationSlotAndPublicationOnTombstone;
89+
}
90+
91+
public void setSkipDropReplicationSlotAndPublicationOnTombstone(
92+
Boolean skipDropReplicationSlotAndPublicationOnTombstone) {
93+
this.skipDropReplicationSlotAndPublicationOnTombstone = skipDropReplicationSlotAndPublicationOnTombstone;
94+
}
95+
8596
public StreamSourcePostgresDebeziumProperties getDebeziumProperties() {
8697
return debeziumProperties;
8798
}

‎stackgres-k8s/src/stream/src/main/java/io/stackgres/stream/jobs/source/AbstractPostgresDebeziumEngineHandler.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -170,8 +170,10 @@ public void handle(boolean success, String message, Throwable error) {
170170
try {
171171
LOGGER.info("Shutting down Event Handler");
172172
eventConsumer.close();
173-
LOGGER.info("Shutting down Debezium Engine");
174-
engine.close();
173+
if (!tombstoneSignalAction.hasTombstoneBeenReceived()) {
174+
LOGGER.info("Shutting down Debezium Engine");
175+
engine.close();
176+
}
175177
executor.shutdown();
176178
executor.awaitTermination(60, TimeUnit.SECONDS);
177179
} catch (Exception shutdownEx) {

‎stackgres-k8s/src/stream/src/main/java/io/stackgres/stream/jobs/source/TombstoneDebeziumSignalAction.java‎

Lines changed: 18 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,10 @@ public boolean arrived(SignalPayload<Partition> signalPayload) throws Interrupte
8585
return false;
8686
}
8787

88+
public synchronized boolean hasTombstoneBeenReceived() {
89+
return tombstoneCompleted != null;
90+
}
91+
8892
public synchronized CompletableFuture<Void> getTombstoneCompleted() {
8993
return tombstoneCompleted != null ? tombstoneCompleted : CompletableFuture.completedFuture(null);
9094
}
@@ -95,6 +99,7 @@ private synchronized void handleTombstone() {
9599
"Tombsone singal can not be handled when stream is already closed or closing");
96100
}
97101
tombstoneCompleted = new CompletableFuture<>();
102+
LOGGER.info("Tombsone singal received");
98103
CompletableFuture
99104
.runAsync(Unchecked.runnable(engine::close))
100105
.handleAsync(Unchecked.biFunction((ignore, ex) -> {
@@ -139,6 +144,7 @@ private void restoreTargetConstraints() {
139144
LOGGER.info("Skipping restoring constraints and indexes for target database on tombstone signal");
140145
return;
141146
}
147+
LOGGER.info("Restoring constraints and indexes for target database on tombstone signal");
142148
final Properties props = new Properties();
143149
final var sgCluster = Optional.of(stream.getSpec().getTarget().getSgCluster());
144150
final String namespace = stream.getMetadata().getNamespace();
@@ -182,7 +188,6 @@ private void restoreTargetConstraints() {
182188
.stream()
183189
.map(e -> Map.entry(e.getKey().toString(), e.getValue().toString()))
184190
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
185-
LOGGER.info("Restoring constraints and indexes for target database on tombstone signal");
186191
try (
187192
SessionFactory sessionFactory = config.getHibernateConfiguration().buildSessionFactory();
188193
StatelessSession session = sessionFactory.openStatelessSession();
@@ -212,6 +217,18 @@ private void restoreTargetConstraints() {
212217
private void cleanupSource() {
213218
if (Objects.equals(stream.getSpec().getSource().getType(), StreamSourceType.SGCLUSTER.toString())
214219
|| Objects.equals(stream.getSpec().getSource().getType(), StreamSourceType.POSTGRES.toString())) {
220+
if (Objects.equals(stream.getSpec().getSource().getType(), StreamSourceType.SGCLUSTER.toString())
221+
&& Optional.of(stream.getSpec().getSource().getSgCluster())
222+
.map(StackGresStreamSourceSgCluster::getSkipDropReplicationSlotAndPublicationOnTombstone)
223+
.orElse(false)
224+
|| Objects.equals(stream.getSpec().getSource().getType(), StreamSourceType.POSTGRES.toString())
225+
&& Optional.of(stream.getSpec().getSource().getPostgres())
226+
.map(StackGresStreamSourcePostgres::getSkipDropReplicationSlotAndPublicationOnTombstone)
227+
.orElse(false)) {
228+
LOGGER.info("Skipping dropping replication slot and publication for source database on tombstone signal");
229+
return;
230+
}
231+
LOGGER.info("Dropping replication slot and publication, if exists, for source database on tombstone signal");
215232
final String slotName = SgClusterDebeziumEngineHandler.slotName(stream);
216233
final String publicationName = SgClusterDebeziumEngineHandler.publicationName(stream);
217234
final Properties props;
@@ -225,7 +242,6 @@ private void cleanupSource() {
225242
.stream()
226243
.map(e -> Map.entry(e.getKey().toString(), e.getValue().toString()))
227244
.collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)));
228-
LOGGER.info("Dropping replication slot and publication, if exists, on tombstone signal");
229245
try (
230246
SessionFactory sessionFactory = config.getHibernateConfiguration().buildSessionFactory();
231247
StatelessSession session = sessionFactory.openStatelessSession();

0 commit comments

Comments
 (0)