refactor: move Kinesis and eventbridge to dynamic properties

Author: Mathieu Gabelle

src/main/java/io/kestra/plugin/aws/eventbridge/PutEvents.java

@@ -10,6 +10,7 @@
 import io.kestra.core.models.executions.metrics.Counter;
 import io.kestra.core.models.executions.metrics.Timer;
 import io.kestra.core.models.flows.State;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.RunnableTask;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.serializers.FileSerde;
@@ -96,14 +97,13 @@ public class PutEvents extends AbstractConnection implements RunnableTask<PutEve
     private static final ObjectMapper MAPPER = JacksonMapper.ofIon()
         .setSerializationInclusion(JsonInclude.Include.ALWAYS);
 
-    @PluginProperty(dynamic = false)
     @NotNull
     @Schema(
         title = "Mark the task as failed when sending an event is unsuccessful.",
         description = "If true, the task will fail when any event fails to be sent."
     )
     @Builder.Default
-    private boolean failOnUnsuccessfulEvents = true;
+    private Property<Boolean> failOnUnsuccessfulEvents = Property.of(true);
 
     @PluginProperty(dynamic = true)
     @NotNull
@@ -129,7 +129,7 @@ public PutEvents.Output run(RunContext runContext) throws Exception {
         runContext.metric(Counter.of("entryCount", entryList.size()));
 
         // Fail if failOnUnsuccessfulEvents
-        if (failOnUnsuccessfulEvents && putEventsResponse.failedEntryCount() > 0) {
+        if (runContext.render(failOnUnsuccessfulEvents).as(Boolean.class).orElseThrow() && putEventsResponse.failedEntryCount() > 0) {
             var logger = runContext.logger();
             logger.error("Response show {} event failed: {}", putEventsResponse.failedEntryCount(), putEventsResponse);
             throw new RuntimeException(String.format("Response show %d event failed: %s", putEventsResponse.failedEntryCount(), putEventsResponse));

src/main/java/io/kestra/plugin/aws/kinesis/PutRecords.java

@@ -11,6 +11,7 @@
 import io.kestra.core.models.executions.metrics.Counter;
 import io.kestra.core.models.executions.metrics.Timer;
 import io.kestra.core.models.flows.State;
+import io.kestra.core.models.property.Property;
 import io.kestra.core.models.tasks.RunnableTask;
 import io.kestra.core.runners.RunContext;
 import io.kestra.core.serializers.FileSerde;
@@ -82,7 +83,7 @@
                     secretKeyId: "<secret-key>"
                     region: "eu-central-1"
                     streamName: "mystream"
-                    records: kestra:///myfile.ion 
+                    records: kestra:///myfile.ion
                 """
         )
     }
@@ -94,28 +95,25 @@ public class PutRecords extends AbstractConnection implements RunnableTask<PutRe
     private static final ObjectMapper MAPPER = JacksonMapper.ofIon()
         .setSerializationInclusion(JsonInclude.Include.ALWAYS);
 
-    @PluginProperty
     @NotNull
     @Schema(
         title = "Mark the task as failed when sending a record is unsuccessful.",
         description = "If true, the task will fail when any record fails to be sent."
     )
     @Builder.Default
-    private boolean failOnUnsuccessfulRecords = true;
+    private Property<Boolean> failOnUnsuccessfulRecords = Property.of(true);
 
-    @PluginProperty(dynamic = true)
     @Schema(
         title = "The name of the stream to push the records.",
         description = "Make sure to set either `streamName` or `streamArn`. One of those must be provided."
     )
-    private String streamName;
+    private Property<String> streamName;
 
-    @PluginProperty(dynamic = true)
     @Schema(
         title = "The ARN of the stream to push the records.",
         description = "Make sure to set either `streamName` or `streamArn`. One of those must be provided."
     )
-    private String streamArn;
+    private Property<String> streamArn;
 
     @PluginProperty(dynamic = true)
     @Schema(
@@ -135,7 +133,7 @@ public Output run(RunContext runContext) throws Exception {
         PutRecordsResponse putRecordsResponse = putRecords(runContext, records);
 
         // Fail if failOnUnsuccessfulRecords
-        if (failOnUnsuccessfulRecords && putRecordsResponse.failedRecordCount() > 0) {
+        if (runContext.render(failOnUnsuccessfulRecords).as(Boolean.class).orElseThrow() && putRecordsResponse.failedRecordCount() > 0) {
             var logger = runContext.logger();
             logger.error("Response show {} record failed: {}", putRecordsResponse.failedRecordCount(), putRecordsResponse);
             throw new RuntimeException(String.format("Response show %d record failed: %s", putRecordsResponse.failedRecordCount(), putRecordsResponse));
@@ -159,10 +157,12 @@ private PutRecordsResponse putRecords(RunContext runContext, List<Record> record
         try (KinesisClient client = client(runContext)) {
             PutRecordsRequest.Builder builder = PutRecordsRequest.builder();
 
-            if (!Strings.isNullOrEmpty(streamArn)) {
-                builder.streamARN(streamArn);
-            } else if (!Strings.isNullOrEmpty(streamName)) {
-                builder.streamName(streamName);
+            var renderedStreamArn = runContext.render(streamArn).as(String.class).orElse(null);
+            var renderedStreamName = runContext.render(streamName).as(String.class).orElse(null);
+            if (!Strings.isNullOrEmpty(renderedStreamArn)) {
+                builder.streamARN(renderedStreamArn);
+            } else if (!Strings.isNullOrEmpty(renderedStreamName)) {
+                builder.streamName(renderedStreamName);
             } else {
                 throw new IllegalArgumentException("Either streamName or streamArn has to be set.");
             }

src/test/java/io/kestra/plugin/aws/kinesis/PutRecordsTest.java

@@ -121,7 +121,7 @@ void runMap() throws Exception {
             .region(Property.of(localstack.getRegion()))
             .accessKeyId(Property.of(localstack.getAccessKey()))
             .secretKeyId(Property.of(localstack.getSecretKey()))
-            .streamName("streamName")
+            .streamName(Property.of("streamName"))
             .records(List.of(record, record2, record3))
             .build();
 
@@ -174,7 +174,7 @@ void runStorage() throws Exception {
             .accessKeyId(Property.of(localstack.getAccessKey()))
             .secretKeyId(Property.of(localstack.getSecretKey()))
             .records(runContext.storage().putFile(tempFile).toString())
-            .streamName("streamName")
+            .streamName(Property.of("streamName"))
             .build();
 
 
@@ -227,7 +227,7 @@ void runStorageUpperCase() throws Exception {
             .accessKeyId(Property.of(localstack.getAccessKey()))
             .secretKeyId(Property.of(localstack.getSecretKey()))
             .records(runContext.storage().putFile(tempFile).toString())
-            .streamName("streamName")
+            .streamName(Property.of("streamName"))
             .build();