Skip to content

Commit 354d5ec

Browse files
teoincontattoteoincontatto
committed
Merge branch 'improve-sgstream-for-incremental-snapshots' into 'main'
SGStream improvements See merge request ongresinc/stackgres!1659
2 parents ff159c4 + fd513c7 commit 354d5ec

21 files changed

Lines changed: 452 additions & 39 deletions

‎stackgres-k8s/src/common/src/main/java/io/stackgres/common/StreamPath.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,8 @@ public enum StreamPath implements EnvVarPathSource<StackGresCluster> {
2222
LOCAL_BIN_PATH("/usr/local/bin"),
2323
DEBEZIUM_BASE_PATH("/var/lib/debezium"),
2424
DEBEZIUM_OFFSET_STORAGE_PATH(DEBEZIUM_BASE_PATH, "offsets.dat"),
25-
DEBEZIUM_DATABASE_HISTORY_PATH(DEBEZIUM_BASE_PATH, "dbhistory.dat");
25+
DEBEZIUM_DATABASE_HISTORY_PATH(DEBEZIUM_BASE_PATH, "dbhistory.dat"),
26+
DEBEZIUM_ANNOTATION_SIGNAL_PATH(DEBEZIUM_BASE_PATH, "annotation-signal.properties");
2627

2728
private final String path;
2829

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

Lines changed: 60 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,16 @@ public class StackGresStreamDebeziumEngineProperties {
4040
@DebeziumMapOptions(generateSummary = true)
4141
private Map<String, Map<String, String>> predicates;
4242

43+
private Integer recordProcessingThreads;
44+
45+
private Integer recordProcessingShutdownTimeoutMs;
46+
47+
private String recordProcessingOrder;
48+
49+
private Boolean recordProcessingWithSerialConsumer;
50+
51+
private Integer taskManagementTimeoutMs;
52+
4353
public String getOffsetCommitPolicy() {
4454
return offsetCommitPolicy;
4555
}
@@ -104,10 +114,52 @@ public void setPredicates(Map<String, Map<String, String>> predicates) {
104114
this.predicates = predicates;
105115
}
106116

117+
public Integer getRecordProcessingThreads() {
118+
return recordProcessingThreads;
119+
}
120+
121+
public void setRecordProcessingThreads(Integer recordProcessingThreads) {
122+
this.recordProcessingThreads = recordProcessingThreads;
123+
}
124+
125+
public Integer getRecordProcessingShutdownTimeoutMs() {
126+
return recordProcessingShutdownTimeoutMs;
127+
}
128+
129+
public void setRecordProcessingShutdownTimeoutMs(Integer recordProcessingShutdownTimeoutMs) {
130+
this.recordProcessingShutdownTimeoutMs = recordProcessingShutdownTimeoutMs;
131+
}
132+
133+
public String getRecordProcessingOrder() {
134+
return recordProcessingOrder;
135+
}
136+
137+
public void setRecordProcessingOrder(String recordProcessingOrder) {
138+
this.recordProcessingOrder = recordProcessingOrder;
139+
}
140+
141+
public Boolean getRecordProcessingWithSerialConsumer() {
142+
return recordProcessingWithSerialConsumer;
143+
}
144+
145+
public void setRecordProcessingWithSerialConsumer(Boolean recordProcessingWithSerialConsumer) {
146+
this.recordProcessingWithSerialConsumer = recordProcessingWithSerialConsumer;
147+
}
148+
149+
public Integer getTaskManagementTimeoutMs() {
150+
return taskManagementTimeoutMs;
151+
}
152+
153+
public void setTaskManagementTimeoutMs(Integer taskManagementTimeoutMs) {
154+
this.taskManagementTimeoutMs = taskManagementTimeoutMs;
155+
}
156+
107157
@Override
108158
public int hashCode() {
109159
return Objects.hash(errorsMaxRetries, errorsRetryDelayInitialMs, errorsRetryDelayMaxMs,
110-
offsetCommitPolicy, offsetFlushIntervalMs, offsetFlushTimeoutMs, predicates, transforms);
160+
offsetCommitPolicy, offsetFlushIntervalMs, offsetFlushTimeoutMs, predicates,
161+
recordProcessingOrder, recordProcessingShutdownTimeoutMs, recordProcessingThreads,
162+
recordProcessingWithSerialConsumer, taskManagementTimeoutMs, transforms);
111163
}
112164

113165
@Override
@@ -126,6 +178,13 @@ public boolean equals(Object obj) {
126178
&& Objects.equals(offsetFlushIntervalMs, other.offsetFlushIntervalMs)
127179
&& Objects.equals(offsetFlushTimeoutMs, other.offsetFlushTimeoutMs)
128180
&& Objects.equals(predicates, other.predicates)
181+
&& Objects.equals(recordProcessingOrder, other.recordProcessingOrder)
182+
&& Objects.equals(recordProcessingShutdownTimeoutMs,
183+
other.recordProcessingShutdownTimeoutMs)
184+
&& Objects.equals(recordProcessingThreads, other.recordProcessingThreads)
185+
&& Objects.equals(recordProcessingWithSerialConsumer,
186+
other.recordProcessingWithSerialConsumer)
187+
&& Objects.equals(taskManagementTimeoutMs, other.taskManagementTimeoutMs)
129188
&& Objects.equals(transforms, other.transforms);
130189
}
131190

‎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/java/io/stackgres/common/crd/sgstream/StackGresStreamSpec.java‎

Lines changed: 14 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,8 @@ public class StackGresStreamSpec {
4141
@Valid
4242
private StackGresStreamDebeziumEngineProperties debeziumEngineProperties;
4343

44+
private Boolean useDebeziumAsyncEngine;
45+
4446
public StackGresStreamSource getSource() {
4547
return source;
4648
}
@@ -82,9 +84,18 @@ public void setDebeziumEngineProperties(
8284
this.debeziumEngineProperties = debeziumEngineProperties;
8385
}
8486

87+
public Boolean getUseDebeziumAsyncEngine() {
88+
return useDebeziumAsyncEngine;
89+
}
90+
91+
public void setUseDebeziumAsyncEngine(Boolean useDebeziumAsyncEngine) {
92+
this.useDebeziumAsyncEngine = useDebeziumAsyncEngine;
93+
}
94+
8595
@Override
8696
public int hashCode() {
87-
return Objects.hash(debeziumEngineProperties, maxRetries, pods, source, target);
97+
return Objects.hash(debeziumEngineProperties, maxRetries, pods, source, target,
98+
useDebeziumAsyncEngine);
8899
}
89100

90101
@Override
@@ -98,7 +109,8 @@ public boolean equals(Object obj) {
98109
StackGresStreamSpec other = (StackGresStreamSpec) obj;
99110
return Objects.equals(debeziumEngineProperties, other.debeziumEngineProperties)
100111
&& Objects.equals(maxRetries, other.maxRetries) && Objects.equals(pods, other.pods)
101-
&& Objects.equals(source, other.source) && Objects.equals(target, other.target);
112+
&& Objects.equals(source, other.source) && Objects.equals(target, other.target)
113+
&& Objects.equals(useDebeziumAsyncEngine, other.useDebeziumAsyncEngine);
102114
}
103115

104116
@Override

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

Lines changed: 23 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -39,6 +39,10 @@ public class StackGresStreamTargetSgCluster {
3939

4040
private String ddlImportRoleSkipFilter;
4141

42+
private Boolean skipDropIndexesAndConstraints;
43+
44+
private Boolean skipRestoreIndexesAfterSnapshot;
45+
4246
@Valid
4347
private StackGresStreamTargetJdbcSinkDebeziumProperties debeziumProperties;
4448

@@ -90,6 +94,22 @@ public void setDdlImportRoleSkipFilter(String ddlImportRoleSkipFilter) {
9094
this.ddlImportRoleSkipFilter = ddlImportRoleSkipFilter;
9195
}
9296

97+
public Boolean getSkipDropIndexesAndConstraints() {
98+
return skipDropIndexesAndConstraints;
99+
}
100+
101+
public void setSkipDropIndexesAndConstraints(Boolean skipDropIndexesAndConstraints) {
102+
this.skipDropIndexesAndConstraints = skipDropIndexesAndConstraints;
103+
}
104+
105+
public Boolean getSkipRestoreIndexesAfterSnapshot() {
106+
return skipRestoreIndexesAfterSnapshot;
107+
}
108+
109+
public void setSkipRestoreIndexesAfterSnapshot(Boolean skipRestoreIndexesAfterSnapshot) {
110+
this.skipRestoreIndexesAfterSnapshot = skipRestoreIndexesAfterSnapshot;
111+
}
112+
93113
public StackGresStreamTargetJdbcSinkDebeziumProperties getDebeziumProperties() {
94114
return debeziumProperties;
95115
}
@@ -102,7 +122,7 @@ public void setDebeziumProperties(
102122
@Override
103123
public int hashCode() {
104124
return Objects.hash(database, ddlImportRoleSkipFilter, debeziumProperties, name, password,
105-
skipDdlImport, username);
125+
skipDdlImport, skipDropIndexesAndConstraints, skipRestoreIndexesAfterSnapshot, username);
106126
}
107127

108128
@Override
@@ -119,6 +139,8 @@ public boolean equals(Object obj) {
119139
&& Objects.equals(debeziumProperties, other.debeziumProperties)
120140
&& Objects.equals(name, other.name) && Objects.equals(password, other.password)
121141
&& Objects.equals(skipDdlImport, other.skipDdlImport)
142+
&& Objects.equals(skipDropIndexesAndConstraints, other.skipDropIndexesAndConstraints)
143+
&& Objects.equals(skipRestoreIndexesAfterSnapshot, other.skipRestoreIndexesAfterSnapshot)
122144
&& Objects.equals(username, other.username);
123145
}
124146

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

Lines changed: 44 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: |
@@ -891,6 +894,14 @@ spec:
891894
Allow to set a [SIMILAR TO regular expression](https://www.postgresql.org/docs/current/functions-matching.html#FUNCTIONS-SIMILARTO-REGEXP) to match the names of the roles to skip during import of DDL.
892895
893896
When not set and source is an SGCluster will match the superuser, replicator and authenticator usernames.
897+
skipDropIndexesAndConstraints:
898+
type: boolean
899+
description: |
900+
When `true` disable drop of indexes and constraints that improve snapshotting performance.
901+
skipRestoreIndexesAfterSnapshot:
902+
type: boolean
903+
description: |
904+
When `true` disable restore of indexes on the first non-snapshot event. This option is required when using incremental snapshotting. This option is ignored when `skipDropIndexesAndConstraints` is set to `true`.
894905
debeziumProperties: &target-jdbc-debeziumProperties
895906
type: object
896907
description: |
@@ -1155,6 +1166,39 @@ spec:
11551166
description: The properties of this predicate.
11561167
additionalProperties:
11571168
type: string
1169+
recordProcessingThreads:
1170+
type: integer
1171+
description: |
1172+
The number of threads that are available to process change event records. If no value is specified (the default), the engine uses the Java [ThreadPoolExecutor](https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/concurrent/ThreadPoolExecutor.html) to dynamically adjust the number of threads, based on the current workload. Maximum number of threads is number of CPU cores on given machine. If a value is specified, the engine uses the Java [fixed thread pool](https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/concurrent/Executors.html) method to create a thread pool with the specified number of threads. To use all available cores on given machine, set the placeholder value, AVAILABLE_CORES.
1173+
recordProcessingShutdownTimeoutMs:
1174+
type: integer
1175+
description: Default `1000`. Maximum time in milliseconds to wait for processing submitted records after a task shutdown is called.
1176+
recordProcessingOrder:
1177+
type: string
1178+
description: |
1179+
Default `ORDERED`.
1180+
1181+
Determines how the records should be produced.
1182+
1183+
* `ORDERED`
1184+
1185+
Records are processed sequentially; that is, they are produced in the order in which they were obtained from the database.
1186+
1187+
* `UNORDERED`
1188+
1189+
Records are processed non-sequentially; that is, they can be produced in an different order than in the source database.
1190+
1191+
The non-sequential processing of the `UNORDERED` option results in better throughput, because records are produced immediately after any SMT processing and message serialization is complete, without waiting for other records. This option doesn’t have any effect when the ChangeConsumer method is provided to the engine.
1192+
recordProcessingWithSerialConsumer:
1193+
type: boolean
1194+
description: Default `false`. Specifies whether the default ChangeConsumer should be created from the provided Consumer, resulting in serial Consumer processing. This option has no effect if you specified the ChangeConsumer interface when you used the API to create the engine.
1195+
taskManagementTimeoutMs:
1196+
type: integer
1197+
description: Default `180000`. Time, in milliseconds, that the engine waits for a task’s lifecycle management operations (starting and stopping) to complete.
1198+
useDebeziumAsyncEngine:
1199+
type: boolean
1200+
description: When `true` use Debezium asyncronous engine. See https://debezium.io/blog/2024/07/08/async-embedded-engine/
1201+
default: true
11581202
status:
11591203
type: object
11601204
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

Lines changed: 5 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -25,13 +25,13 @@
2525

2626
@Singleton
2727
@ValidationType(ErrorType.FORBIDDEN_STREAM_UPDATE)
28-
public class LockValidator implements StreamValidator {
28+
public class StatusLockValidator implements StreamValidator {
2929

3030
final ObjectMapper objectMapper;
3131
final int duration;
3232

3333
@Inject
34-
public LockValidator(OperatorPropertyContext operatorPropertyContext,
34+
public StatusLockValidator(OperatorPropertyContext operatorPropertyContext,
3535
ObjectMapper objectMapper) {
3636
this.duration = operatorPropertyContext.getInt(OperatorProperty.LOCK_DURATION);
3737
this.objectMapper = objectMapper;
@@ -45,8 +45,8 @@ public void validate(StackGresStreamReview review) throws ValidationFailed {
4545
case UPDATE: {
4646
StackGresStream stream = review.getRequest().getObject();
4747
StackGresStream oldStream = review.getRequest().getOldObject();
48-
if (Objects.equals(objectMapper.valueToTree(stream.getSpec()),
49-
objectMapper.valueToTree(oldStream.getSpec()))) {
48+
if (Objects.equals(objectMapper.valueToTree(stream.getStatus()),
49+
objectMapper.valueToTree(oldStream.getStatus()))) {
5050
return;
5151
}
5252
String username = review.getRequest().getUserInfo().getUsername();
@@ -59,7 +59,7 @@ public void validate(StackGresStreamReview review) throws ValidationFailed {
5959
getServiceAccountFromUsername(username))
6060
)
6161
) {
62-
fail("SGStream update is forbidden. It is locked by the SGStream"
62+
fail("SGStream status update is forbidden. It is locked by the SGStream"
6363
+ " that is currently running. Please, wait for the operation to finish,"
6464
+ " stop the operation by deleting it or wait for the lock duration of "
6565
+ duration + " seconds to expire.");

‎stackgres-k8s/src/operator/src/main/java/io/stackgres/operator/validation/stream/StreamImmutableSpecValidator.java‎

Lines changed: 6 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,8 @@
55

66
package io.stackgres.operator.validation.stream;
77

8+
import java.util.Optional;
9+
810
import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
911
import io.stackgres.common.ErrorType;
1012
import io.stackgres.common.crd.sgstream.StackGresStream;
@@ -26,8 +28,10 @@ public void validate(StackGresStreamReview review) throws ValidationFailed {
2628
case UPDATE:
2729
StackGresStream stream = review.getRequest().getObject();
2830
StackGresStream oldStream = review.getRequest().getOldObject();
29-
if (!stream.getSpec().equals(oldStream.getSpec())) {
30-
fail("spec can not be updated");
31+
if (Optional.ofNullable(stream.getSpec().getMaxRetries()).orElse(-1) >= 0) {
32+
if (!stream.getSpec().equals(oldStream.getSpec())) {
33+
fail("spec can not be updated when maxRetries is greater than -1");
34+
}
3135
}
3236
break;
3337
default:

0 commit comments

Comments
 (0)