refactor: migrate to dynamic properties (#81)

* refactor: migrate to dynamic properties

migrate Abstract

Postgres properties

MySQL properties

DB2, Oracle properties

mongodb properties

SQLServer properties

Author: mgabelle

plugin-debezium-db2/src/main/java/io/kestra/plugin/debezium/db2/Capture.java

@@ -3,6 +3,7 @@
 import io.debezium.connector.db2.Db2Connector;
 import io.kestra.core.models.annotations.Example;
 import io.kestra.core.models.annotations.Plugin;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.runners.RunContext;
 import io.kestra.plugin.debezium.AbstractDebeziumTask;
 import io.swagger.v3.oas.annotations.media.Schema;
@@ -38,10 +39,10 @@
 )
 public class Capture extends AbstractDebeziumTask implements Db2Interface {
 
-    protected String database;
+    protected Property<String> database;
 
     @Builder.Default
-    private Db2Interface.SnapshotMode snapshotMode = Db2Interface.SnapshotMode.INITIAL;
+    private Property<Db2Interface.SnapshotMode> snapshotMode = Property.of(SnapshotMode.INITIAL);
 
     @Override
     protected boolean needDatabaseHistory() {
@@ -54,12 +55,12 @@ protected Properties properties(RunContext runContext, Path offsetFile, Path his
 
         props.setProperty("connector.class", Db2Connector.class.getName());
 
-        props.setProperty("database.dbname", runContext.render(this.database));
+        props.setProperty("database.dbname", runContext.render(this.database).as(String.class).orElseThrow());
 
         props.setProperty("include.schema.changes", "false");
 
         if (this.snapshotMode != null) {
-            props.setProperty("snapshot.mode", this.snapshotMode.name().toLowerCase(Locale.ROOT));
+            props.setProperty("snapshot.mode", runContext.render(this.snapshotMode).as(SnapshotMode.class).orElseThrow().name().toLowerCase(Locale.ROOT));
         }
 
         return props;

plugin-debezium-db2/src/main/java/io/kestra/plugin/debezium/db2/Db2Interface.java

@@ -1,6 +1,7 @@
 package io.kestra.plugin.debezium.db2;
 
 import io.kestra.core.models.annotations.PluginProperty;
+import io.kestra.core.models.property.Property;
 import io.swagger.v3.oas.annotations.media.Schema;
 
 import jakarta.validation.constraints.NotNull;
@@ -9,9 +10,8 @@ public interface Db2Interface {
     @Schema(
         title = "The name of the DB2 database from which to stream the changes."
     )
-    @PluginProperty(dynamic = true)
     @NotNull
-    String getDatabase();
+    Property<String> getDatabase();
 
     @Schema(
         title = "Specifies the criteria for running a snapshot when the connector starts.",
@@ -23,9 +23,8 @@ public interface Db2Interface {
             "- `NO_DATA`: The connector captures the structure of all relevant tables, performing all the steps described in the INITIAL, except that it does not create READ events to represent the data set at the point of the connector’s start-up.\n" +
             "- `RECOVERY`: Set this option to restore a database schema history topic that is lost or corrupted. After a restart, the connector runs a snapshot that rebuilds the topic from the source tables."
     )
-    @PluginProperty(dynamic = false)
     @NotNull
-    SnapshotMode getSnapshotMode();
+    Property<SnapshotMode> getSnapshotMode();
 
     public enum SnapshotMode {
         ALWAYS,

plugin-debezium-db2/src/main/java/io/kestra/plugin/debezium/db2/RealtimeTrigger.java

@@ -4,6 +4,7 @@
 import io.kestra.core.models.annotations.Plugin;
 import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.triggers.*;
 import io.kestra.plugin.debezium.AbstractDebeziumInterface;
 import io.kestra.plugin.debezium.AbstractDebeziumRealtimeTrigger;
@@ -50,9 +51,9 @@
 )
 public class RealtimeTrigger extends AbstractDebeziumRealtimeTrigger implements Db2Interface, AbstractDebeziumInterface {
     @Builder.Default
-    private SnapshotMode snapshotMode = SnapshotMode.INITIAL;
+    private Property<SnapshotMode> snapshotMode = Property.of(SnapshotMode.INITIAL);
 
-    private String database;
+    private Property<String> database;
 
     @Override
     public Publisher<Execution> evaluate(ConditionContext conditionContext, TriggerContext context) throws Exception {

plugin-debezium-db2/src/main/java/io/kestra/plugin/debezium/db2/Trigger.java

@@ -4,6 +4,7 @@
 import io.kestra.core.models.annotations.Plugin;
 import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.triggers.*;
 import io.kestra.core.runners.RunContext;
 import io.kestra.plugin.debezium.AbstractDebeziumInterface;
@@ -42,9 +43,9 @@
 )
 public class Trigger extends AbstractDebeziumTrigger implements Db2Interface, AbstractDebeziumInterface {
     @Builder.Default
-    private Db2Interface.SnapshotMode snapshotMode = Db2Interface.SnapshotMode.INITIAL;
+    private Property<SnapshotMode> snapshotMode = Property.of(SnapshotMode.INITIAL);
 
-    private String database;
+    private Property<String> database;
 
     @Override
     public Optional<Execution> evaluate(ConditionContext conditionContext, TriggerContext context) throws Exception {

plugin-debezium-db2/src/test/java/io/kestra/plugin/debezium/db2/CaptureTest.java

@@ -1,6 +1,7 @@
 package io.kestra.plugin.debezium.db2;
 
 import com.google.common.base.Charsets;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.serializers.FileSerde;
 import io.kestra.core.storages.StorageInterface;
 import io.kestra.core.utils.IdUtils;
@@ -59,13 +60,13 @@ void run() throws Exception {
         Capture task = Capture.builder()
             .id(IdUtils.create())
             .type(Capture.class.getName())
-            .snapshotMode(Db2Interface.SnapshotMode.INITIAL)
-            .hostname("127.0.0.1")
-            .port("5023")
-            .username(getUsername())
-            .password(getPassword())
-            .database("kestra")
-            .maxRecords(5)
+            .snapshotMode(Property.of(Db2Interface.SnapshotMode.INITIAL))
+            .hostname(Property.of("127.0.0.1"))
+            .port(Property.of("5023"))
+            .username(Property.of(getUsername()))
+            .password(Property.of(getPassword()))
+            .database(Property.of("kestra"))
+            .maxRecords(Property.of(5))
             .includedTables(List.of("DB2INST1.EVENTS"))
             .build();
 

plugin-debezium-mongodb/src/main/java/io/kestra/plugin/debezium/mongodb/Capture.java

@@ -3,6 +3,7 @@
 import io.debezium.connector.mongodb.MongoDbConnector;
 import io.kestra.core.models.annotations.Example;
 import io.kestra.core.models.annotations.Plugin;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.runners.RunContext;
 import io.kestra.plugin.debezium.AbstractDebeziumTask;
 import io.swagger.v3.oas.annotations.media.Schema;
@@ -65,10 +66,10 @@ public class Capture extends AbstractDebeziumTask implements MongodbInterface {
     private Object excludedCollections;
 
     @NotNull
-    private String connectionString;
+    private Property<String> connectionString;
 
     @Builder.Default
-    private MongodbInterface.SnapshotMode snapshotMode = MongodbInterface.SnapshotMode.INITIAL;
+    private Property<MongodbInterface.SnapshotMode> snapshotMode = Property.of(SnapshotMode.INITIAL);
 
     @Override
     protected boolean needDatabaseHistory() {
@@ -81,7 +82,7 @@ protected Properties properties(RunContext runContext, Path offsetFile, Path his
 
         props.setProperty("connector.class", MongoDbConnector.class.getName());
 
-        props.setProperty("mongodb.connection.string", runContext.render(this.connectionString));
+        props.setProperty("mongodb.connection.string", runContext.render(this.connectionString).as(String.class).orElse(null));
 
         if (this.includedCollections != null) {
             props.setProperty("collection.include.list", joinProperties(runContext, this.includedCollections));
@@ -94,7 +95,7 @@ protected Properties properties(RunContext runContext, Path offsetFile, Path his
         props.setProperty("capture.mode", "change_streams_update_full_with_pre_image");
 
         if (this.snapshotMode != null) {
-            props.setProperty("snapshot.mode", this.snapshotMode.name().toLowerCase(Locale.ROOT));
+            props.setProperty("snapshot.mode", runContext.render(this.snapshotMode).as(SnapshotMode.class).orElseThrow().name().toLowerCase(Locale.ROOT));
         }
 
         return props;

plugin-debezium-mongodb/src/main/java/io/kestra/plugin/debezium/mongodb/MongodbInterface.java

@@ -1,6 +1,7 @@
 package io.kestra.plugin.debezium.mongodb;
 
 import io.kestra.core.models.annotations.PluginProperty;
+import io.kestra.core.models.property.Property;
 import io.swagger.v3.oas.annotations.media.Schema;
 import jakarta.validation.constraints.NotNull;
 import lombok.Getter;
@@ -15,8 +16,7 @@ public interface MongodbInterface {
             "mongodb://mongo_user:mongo_passwd@mongos0.example.com:27017,mongos1.example.com:27017/"
         }
     )
-    @PluginProperty(dynamic = true)
-    String getConnectionString();
+    Property<String> getConnectionString();
 
     @Schema(
         title = "The name of the MongoDB database collection included from which to stream the changes.",
@@ -42,9 +42,8 @@ public interface MongodbInterface {
             "- `NO_DATA`: The connector captures the structure of all relevant tables, performing all the steps described in the default snapshot workflow, except that it does not create READ events to represent the data set at the point of the connector’s start-up.\n" +
             "- `WHEN_NEEDED`: The connector runs a snapshot upon startup whenever it deems it necessary. That is, when no offsets are available, or when a previously recorded offset specifies a binlog location or GTID that is not available in the server."
     )
-    @PluginProperty(dynamic = false)
     @NotNull
-    SnapshotMode getSnapshotMode();
+    Property<SnapshotMode> getSnapshotMode();
 
     public enum SnapshotMode {
         INITIAL,

plugin-debezium-mongodb/src/main/java/io/kestra/plugin/debezium/mongodb/RealtimeTrigger.java

@@ -4,6 +4,7 @@
 import io.kestra.core.models.annotations.Plugin;
 import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.triggers.TriggerContext;
 import io.kestra.core.models.triggers.TriggerService;
 import io.kestra.plugin.debezium.AbstractDebeziumInterface;
@@ -69,14 +70,14 @@
 )
 public class RealtimeTrigger extends AbstractDebeziumRealtimeTrigger implements MongodbInterface, AbstractDebeziumInterface {
     @Builder.Default
-    private MongodbInterface.SnapshotMode snapshotMode = MongodbInterface.SnapshotMode.INITIAL;
+    private Property<MongodbInterface.SnapshotMode> snapshotMode = Property.of(SnapshotMode.INITIAL);
 
     private Object includedCollections;
 
     private Object excludedCollections;
 
     @NotNull
-    private String connectionString;
+    private Property<String> connectionString;
 
     @Override
     public Publisher<Execution> evaluate(ConditionContext conditionContext, TriggerContext context) throws Exception {

plugin-debezium-mongodb/src/main/java/io/kestra/plugin/debezium/mongodb/Trigger.java

@@ -4,6 +4,7 @@
 import io.kestra.core.models.annotations.Plugin;
 import io.kestra.core.models.conditions.ConditionContext;
 import io.kestra.core.models.executions.Execution;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.triggers.*;
 import io.kestra.core.runners.RunContext;
 import io.kestra.plugin.debezium.AbstractDebeziumInterface;
@@ -75,41 +76,41 @@ public class Trigger extends AbstractTrigger implements PollingTriggerInterface,
     private final Duration interval = Duration.ofSeconds(60);
 
     @Builder.Default
-    protected AbstractDebeziumTask.Format format = AbstractDebeziumTask.Format.INLINE;
+    protected Property<AbstractDebeziumTask.Format> format = Property.of(AbstractDebeziumTask.Format.INLINE);
 
     @Builder.Default
-    protected AbstractDebeziumTask.Deleted deleted = AbstractDebeziumTask.Deleted.ADD_FIELD;
+    protected Property<AbstractDebeziumTask.Deleted> deleted = Property.of(AbstractDebeziumTask.Deleted.ADD_FIELD);
 
     @Builder.Default
-    protected String deletedFieldName = "deleted";
+    protected Property<String> deletedFieldName = Property.of("deleted");
 
     @Builder.Default
-    protected AbstractDebeziumTask.Key key = AbstractDebeziumTask.Key.ADD_FIELD;
+    protected Property<AbstractDebeziumTask.Key> key = Property.of(AbstractDebeziumTask.Key.ADD_FIELD);
 
     @Builder.Default
-    protected AbstractDebeziumTask.Metadata metadata = AbstractDebeziumTask.Metadata.ADD_FIELD;
+    protected Property<AbstractDebeziumTask.Metadata> metadata = Property.of(AbstractDebeziumTask.Metadata.ADD_FIELD);
 
     @Builder.Default
-    protected String metadataFieldName = "metadata";
+    protected Property<String> metadataFieldName = Property.of("metadata");
 
     @Builder.Default
-    protected AbstractDebeziumTask.SplitTable splitTable = AbstractDebeziumTask.SplitTable.TABLE;
+    protected Property<AbstractDebeziumTask.SplitTable> splitTable = Property.of(AbstractDebeziumTask.SplitTable.TABLE);
 
     @Builder.Default
-    protected Boolean ignoreDdl = true;
+    protected Property<Boolean> ignoreDdl = Property.of(true);
 
     @NotNull
-    private String connectionString;
+    private Property<String> connectionString;
 
     @Builder.Default
-    protected String hostname = "";
+    protected Property<String> hostname = Property.of("");
 
     @Builder.Default
-    protected String port = "";
+    protected Property<String> port = Property.of("");
 
-    protected String username;
+    protected Property<String> username;
 
-    protected String password;
+    protected Property<String> password;
 
     private Object includedDatabases;
 
@@ -127,20 +128,20 @@ public class Trigger extends AbstractTrigger implements PollingTriggerInterface,
 
     private Object excludedColumns;
 
-    private Map<String, String> properties;
+    private Property<Map<String, String>> properties;
 
     @Builder.Default
-    protected String stateName = "debezium-state";
+    protected Property<String> stateName = Property.of("debezium-state");
 
-    private Integer maxRecords;
+    private Property<Integer> maxRecords;
 
-    private Duration maxDuration;
+    private Property<Duration> maxDuration;
 
     @Builder.Default
-    private Duration maxWait = Duration.ofSeconds(10);
+    private Property<Duration> maxWait = Property.of(Duration.ofSeconds(10));
 
     @Builder.Default
-    private MongodbInterface.SnapshotMode snapshotMode = MongodbInterface.SnapshotMode.INITIAL;
+    private Property<MongodbInterface.SnapshotMode> snapshotMode = Property.of(SnapshotMode.INITIAL);
 
     @Override
     public Optional<Execution> evaluate(ConditionContext conditionContext, TriggerContext context) throws Exception {

plugin-debezium-mongodb/src/test/java/io/kestra/plugin/debezium/mongodb/CaptureTest.java

@@ -1,6 +1,7 @@
 package io.kestra.plugin.debezium.mongodb;
 
 import com.google.common.base.Charsets;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.serializers.FileSerde;
 import io.kestra.core.storages.StorageInterface;
 import io.kestra.core.utils.IdUtils;
@@ -36,9 +37,9 @@ void run() throws Exception {
         Capture task = Capture.builder()
             .id(IdUtils.create())
             .type(Capture.class.getName())
-            .snapshotMode(MongodbInterface.SnapshotMode.INITIAL)
-            .connectionString("mongodb://mongo_user:mongo_passwd@127.0.0.1:27017/?replicaSet=rs0")
-            .maxRecords(20)
+            .snapshotMode(Property.of(MongodbInterface.SnapshotMode.INITIAL))
+            .connectionString(Property.of("mongodb://mongo_user:mongo_passwd@127.0.0.1:27017/?replicaSet=rs0"))
+            .maxRecords(Property.of(20))
             .build();
 
         RunContext runContext = TestsUtils.mockRunContext(runContextFactory, task, Map.of());