-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
DBZ-8512 Support storages supported by Debezium operator for pipeline…
… in Debezium platform (#8) * DBZ-8512 Use JDBC store as default for offsets and schema history store * DBZ-8512 Enable offset store in helm * DBZ-8512 Configure watcher to create publication only for the outbox table * DBZ-8512 Use CustomStore to configure the DS resources * DBZ-8512 Update debezium operator chart --------- Signed-off-by: Fiore Mario Vitale <[email protected]>
- Loading branch information
Showing
17 changed files
with
605 additions
and
52 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
15 changes: 15 additions & 0 deletions
15
debezium-platform-conductor/src/main/java/io/debezium/platform/config/OffsetConfigGroup.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,15 @@ | ||
| /* | ||
| * Copyright Debezium Authors. | ||
| * | ||
| * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 | ||
| */ | ||
| package io.debezium.platform.config; | ||
|
|
||
| import java.util.Map; | ||
|
|
||
| public interface OffsetConfigGroup { | ||
|
|
||
| OffsetStorageConfigGroup storage(); | ||
|
|
||
| Map<String, String> config(); | ||
| } |
14 changes: 14 additions & 0 deletions
14
...latform-conductor/src/main/java/io/debezium/platform/config/OffsetStorageConfigGroup.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,14 @@ | ||
| /* | ||
| * Copyright Debezium Authors. | ||
| * | ||
| * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 | ||
| */ | ||
| package io.debezium.platform.config; | ||
|
|
||
| import java.util.Map; | ||
|
|
||
| public interface OffsetStorageConfigGroup { | ||
| String type(); | ||
|
|
||
| Map<String, String> config(); | ||
| } |
22 changes: 22 additions & 0 deletions
22
...ium-platform-conductor/src/main/java/io/debezium/platform/config/PipelineConfigGroup.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,22 @@ | ||
| /* | ||
| * Copyright Debezium Authors. | ||
| * | ||
| * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 | ||
| */ | ||
| package io.debezium.platform.config; | ||
|
|
||
| import io.quarkus.runtime.annotations.ConfigPhase; | ||
| import io.quarkus.runtime.annotations.ConfigRoot; | ||
| import io.smallrye.config.ConfigMapping; | ||
| import io.smallrye.config.WithName; | ||
|
|
||
| @ConfigMapping(prefix = "pipeline") | ||
| @ConfigRoot(prefix = "pipeline", phase = ConfigPhase.RUN_TIME) | ||
| public interface PipelineConfigGroup { | ||
|
|
||
| OffsetConfigGroup offset(); | ||
|
|
||
| @WithName("schema.history") | ||
| SchemaHistoryConfigGroup schema(); | ||
|
|
||
| } |
15 changes: 15 additions & 0 deletions
15
...latform-conductor/src/main/java/io/debezium/platform/config/SchemaHistoryConfigGroup.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,15 @@ | ||
| /* | ||
| * Copyright Debezium Authors. | ||
| * | ||
| * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 | ||
| */ | ||
| package io.debezium.platform.config; | ||
|
|
||
| import java.util.Map; | ||
|
|
||
| public interface SchemaHistoryConfigGroup { | ||
|
|
||
| String internal(); | ||
|
|
||
| Map<String, String> config(); | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
82 changes: 82 additions & 0 deletions
82
.../main/java/io/debezium/platform/environment/operator/configuration/TableNameResolver.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,82 @@ | ||
| /* | ||
| * Copyright Debezium Authors. | ||
| * | ||
| * Licensed under the Apache Software License version 2.0, available at http://www.apache.org/licenses/LICENSE-2.0 | ||
| */ | ||
| package io.debezium.platform.environment.operator.configuration; | ||
|
|
||
| import java.util.List; | ||
| import java.util.function.Function; | ||
|
|
||
| import jakarta.enterprise.context.Dependent; | ||
|
|
||
| import io.debezium.platform.domain.views.flat.PipelineFlat; | ||
|
|
||
| @Dependent | ||
| public class TableNameResolver { | ||
|
|
||
| private static final List<PlaceHolder> PLACE_HOLDERS = List.of( | ||
| new PlaceHolder("@{pipeline_name}", PipelineFlat::getName)); | ||
|
|
||
| public String resolve(PipelineFlat pipeline, String currentValue) { | ||
|
|
||
| if (currentValue == null || currentValue.isEmpty()) { | ||
| return currentValue; | ||
| } | ||
|
|
||
| String processedValue = currentValue; | ||
| for (PlaceHolder placeHolder : PLACE_HOLDERS) { | ||
| processedValue = placeHolder.apply(processedValue, pipeline); | ||
| } | ||
|
|
||
| return sanitizeTableName(processedValue); | ||
| } | ||
|
|
||
| private record PlaceHolder(String token, Function<PipelineFlat, String> valueResolver) { | ||
|
|
||
| String apply(String text, PipelineFlat pipeline) { | ||
| return text.replace(token, valueResolver.apply(pipeline)); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Sanitizes a string to be used as a PostgreSQL table name. | ||
| * - Replaces invalid characters with underscores | ||
| * - Ensures the name starts with a letter or underscore | ||
| * - Truncates the name to 63 bytes (PostgreSQL's limit) | ||
| * | ||
| * @param tableName The original table name | ||
| * @return A sanitized version safe for use as a PostgreSQL table name | ||
| */ | ||
| private String sanitizeTableName(String tableName) { | ||
|
|
||
| if (tableName == null || tableName.isEmpty()) { | ||
| throw new IllegalArgumentException("Table name cannot be null or empty"); | ||
| } | ||
|
|
||
| // PostgreSQL folds unquoted identifiers to lowercase | ||
| String sanitized = tableName.toLowerCase(); | ||
|
|
||
| sanitized = sanitized.replaceAll("[^a-z0-9_]", "_"); | ||
|
|
||
| // Ensure the name starts with a letter or underscore | ||
| if (!sanitized.matches("^[a-z_].*")) { | ||
| sanitized = "_" + sanitized; | ||
| } | ||
|
|
||
| // Remove consecutive underscores | ||
| sanitized = sanitized.replaceAll("_+", "_"); | ||
|
|
||
| // Trim trailing underscore | ||
| sanitized = sanitized.replaceAll("_$", ""); | ||
|
|
||
| // Truncate to PostgreSQL's limit of 63 bytes | ||
| if (sanitized.length() > 63) { | ||
| sanitized = sanitized.substring(0, 63); | ||
| // If we happened to end with an underscore after truncating, remove it | ||
| sanitized = sanitized.replaceAll("_$", ""); | ||
| } | ||
|
|
||
| return sanitized; | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.