Skip to content

Commit 69726e4

Browse files
teoincontattoteoincontatto
committed
feat: support for Debezium asyncronous engine
1 parent a45a3e7 commit 69726e4

6 files changed

Lines changed: 181 additions & 4 deletions

File tree

‎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/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/resources/crds/SGStream.yaml‎

Lines changed: 33 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1163,6 +1163,39 @@ spec:
11631163
description: The properties of this predicate.
11641164
additionalProperties:
11651165
type: string
1166+
recordProcessingThreads:
1167+
type: integer
1168+
description: |
1169+
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.
1170+
recordProcessingShutdownTimeoutMs:
1171+
type: integer
1172+
description: Default `1000`. Maximum time in milliseconds to wait for processing submitted records after a task shutdown is called.
1173+
recordProcessingOrder:
1174+
type: string
1175+
description: |
1176+
Default `ORDERED`.
1177+
1178+
Determines how the records should be produced.
1179+
1180+
* `ORDERED`
1181+
1182+
Records are processed sequentially; that is, they are produced in the order in which they were obtained from the database.
1183+
1184+
* `UNORDERED`
1185+
1186+
Records are processed non-sequentially; that is, they can be produced in an different order than in the source database.
1187+
1188+
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.
1189+
recordProcessingWithSerialConsumer:
1190+
type: boolean
1191+
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.
1192+
taskManagementTimeoutMs:
1193+
type: integer
1194+
description: Default `180000`. Time, in milliseconds, that the engine waits for a task’s lifecycle management operations (starting and stopping) to complete.
1195+
useDebeziumAsyncEngine:
1196+
type: boolean
1197+
description: When `true` use Debezium asyncronous engine. See https://debezium.io/blog/2024/07/08/async-embedded-engine/
1198+
default: true
11661199
status:
11671200
type: object
11681201
description: |

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

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,16 @@ public class StreamDebeziumEngineProperties {
3131

3232
private Map<String, Map<String, String>> predicates;
3333

34+
private Integer recordProcessingThreads;
35+
36+
private Integer recordProcessingShutdownTimeoutMs;
37+
38+
private String recordProcessingOrder;
39+
40+
private Boolean recordProcessingWithSerialConsumer;
41+
42+
private Integer taskManagementTimeoutMs;
43+
3444
public String getOffsetCommitPolicy() {
3545
return offsetCommitPolicy;
3646
}
@@ -95,6 +105,46 @@ public void setPredicates(Map<String, Map<String, String>> predicates) {
95105
this.predicates = predicates;
96106
}
97107

108+
public Integer getRecordProcessingThreads() {
109+
return recordProcessingThreads;
110+
}
111+
112+
public void setRecordProcessingThreads(Integer recordProcessingThreads) {
113+
this.recordProcessingThreads = recordProcessingThreads;
114+
}
115+
116+
public Integer getRecordProcessingShutdownTimeoutMs() {
117+
return recordProcessingShutdownTimeoutMs;
118+
}
119+
120+
public void setRecordProcessingShutdownTimeoutMs(Integer recordProcessingShutdownTimeoutMs) {
121+
this.recordProcessingShutdownTimeoutMs = recordProcessingShutdownTimeoutMs;
122+
}
123+
124+
public String getRecordProcessingOrder() {
125+
return recordProcessingOrder;
126+
}
127+
128+
public void setRecordProcessingOrder(String recordProcessingOrder) {
129+
this.recordProcessingOrder = recordProcessingOrder;
130+
}
131+
132+
public Boolean getRecordProcessingWithSerialConsumer() {
133+
return recordProcessingWithSerialConsumer;
134+
}
135+
136+
public void setRecordProcessingWithSerialConsumer(Boolean recordProcessingWithSerialConsumer) {
137+
this.recordProcessingWithSerialConsumer = recordProcessingWithSerialConsumer;
138+
}
139+
140+
public Integer getTaskManagementTimeoutMs() {
141+
return taskManagementTimeoutMs;
142+
}
143+
144+
public void setTaskManagementTimeoutMs(Integer taskManagementTimeoutMs) {
145+
this.taskManagementTimeoutMs = taskManagementTimeoutMs;
146+
}
147+
98148
@Override
99149
public String toString() {
100150
return StackGresUtil.toPrettyYaml(this);

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

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,8 @@ public class StreamSpec {
2323

2424
private StreamDebeziumEngineProperties debeziumEngineProperties;
2525

26+
private Boolean useDebeziumAsyncEngine;
27+
2628
public StreamSource getSource() {
2729
return source;
2830
}
@@ -64,6 +66,14 @@ public void setDebeziumEngineProperties(
6466
this.debeziumEngineProperties = debeziumEngineProperties;
6567
}
6668

69+
public Boolean getUseDebeziumAsyncEngine() {
70+
return useDebeziumAsyncEngine;
71+
}
72+
73+
public void setUseDebeziumAsyncEngine(Boolean useDebeziumAsyncEngine) {
74+
this.useDebeziumAsyncEngine = useDebeziumAsyncEngine;
75+
}
76+
6777
@Override
6878
public String toString() {
6979
return StackGresUtil.toPrettyYaml(this);

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

Lines changed: 14 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -13,9 +13,12 @@
1313
import java.util.concurrent.Executors;
1414
import java.util.concurrent.TimeUnit;
1515

16+
import io.debezium.embedded.ConvertingEngineBuilderFactory;
17+
import io.debezium.embedded.async.ConvertingAsyncEngineBuilderFactory;
1618
import io.debezium.engine.DebeziumEngine;
1719
import io.debezium.engine.DebeziumEngine.CompletionCallback;
1820
import io.debezium.engine.DebeziumEngine.ConnectorCallback;
21+
import io.debezium.engine.format.KeyValueHeaderChangeEventFormat;
1922
import io.debezium.engine.format.SerializationFormat;
2023
import io.fabric8.kubernetes.api.model.Secret;
2124
import io.stackgres.common.StreamPath;
@@ -118,9 +121,19 @@ public <T> CompletableFuture<Void> streamChangeEvents(
118121
setSourceProperties(stream, props);
119122

120123
final CompletableFuture<Void> streamCompleted = new CompletableFuture<>();
124+
final String engineBuilderFactory;
125+
if (Optional.ofNullable(stream.getSpec().getUseDebeziumAsyncEngine())
126+
.orElse(true)) {
127+
engineBuilderFactory = ConvertingAsyncEngineBuilderFactory.class.getName();
128+
} else {
129+
engineBuilderFactory = ConvertingEngineBuilderFactory.class.getName();
130+
}
121131
final DebeziumEngine<?> engine;
122132
try {
123-
engine = DebeziumEngine.create(format)
133+
engine = DebeziumEngine
134+
.create(
135+
KeyValueHeaderChangeEventFormat.of(format, format, format),
136+
engineBuilderFactory)
124137
.using(props)
125138
.using(new ConnectorCallback() {
126139
})

0 commit comments

Comments
 (0)