feat(plugins): add new plugin tasks for EventHubs (#38) #56

fhussonnois · 2024-01-10T16:35:01Z

This commit adds the following tasks types:

io.kesta.plugin.azure.messaging.eventhubs.Produce
io.kesta.plugin.azure.messaging.eventhubs.Consume
io.kesta.plugin.azure.messaging.eventhubs.Trigger

What changes are being made and why?

How the changes have been QAed?

Setup Instructions

anna-geller · 2024-01-12T13:16:20Z

Nice! I see you implemented functionality to consume messages from the EventHubs. Could that be implemented as a trigger in a similar fashion as we support SQS, PubSub or Kafka triggers?

loicmathieu

Didn't look everything but there is a lot of interfaces/asbtract classes that are only implemented once, it would be great to remove them to avoid too many classes if they are not that much important.

Also, we have some package naming convention, please move all your classes from io.kestra.plugin.azure.messaging.eventhubs to io.kestra.plugin.azure.eventhubs;.

io.kestra.plugin.azure id the Azure group of plugins and io.kestra.plugin.azure.eventhubs the subgroup of plugins, we didn't use anymore hierarchy and as the FQCN of the task is used it's important not to have too long packages.

build.gradle

src/main/java/io/kestra/plugin/azure/messaging/eventhubs/config/BlobContainerClientConfig.java

src/main/java/io/kestra/plugin/azure/messaging/eventhubs/client/AzureAbstractClientFactory.java

...in/java/io/kestra/plugin/azure/messaging/eventhubs/service/consumer/CheckpointStoreType.java

loicmathieu · 2024-01-12T14:51:47Z

src/main/java/io/kestra/plugin/azure/messaging/eventhubs/Produce.java

+        InputStream is = null;
+        if (this.getFrom() instanceof String data) {
+            is = reader.read(data);
+        }
+
+        if (this.getFrom() instanceof Map data) {
+            is = reader.read(data);
+        }
+
+        if (this.getFrom() instanceof List data) {
+            is = reader.read(data);
+        }


We usually produce one message by 'line' as we handle data in micro-batch.
If it's a List: one message by element.
If it's a Map, we considered it is an object so a single message.
If it's a String, we consider it is an ION file so one message by line. In this case you should use the FileSerde.read() to read lines and read it in a reactive manner to avoid mounting everything in memory in case the file is big.
You can have a look at an example here: https://github.com/kestra-io/plugin-aws/blob/master/src/main/java/io/kestra/plugin/aws/sns/Publish.java#L68

Yes, basically the InputDataReader is only responsible for providing an IntputStream of ION file whatever the type of input.
Then the EventHubProducerService uses a reactive approach to read that file and produce one event per line.

loicmathieu

Two important points:

All plugin properties must be annotated with @PluginProperty even if not setting any attributes or they will not be properly documented.
All logs are stored in database even if not displayed on the UI so you must restrict for too much logging. I think you can remove some logs

Speaking of logs, you sometimes use if(log.isLevelEnabled) but we're using SLF4J that didn't instanciate strings if the level is not enabled so unless you're competing an expansive parameter you should not do that.

src/main/java/io/kestra/plugin/azure/client/DynamicAzureClientConfig.java

src/main/java/io/kestra/plugin/azure/eventhubs/model/EventDataObject.java

src/main/java/io/kestra/plugin/azure/client/AzureClientConfig.java

src/main/java/io/kestra/plugin/azure/eventhubs/service/consumer/EventHubConsumerService.java

src/main/java/io/kestra/plugin/azure/eventhubs/Trigger.java

loicmathieu · 2024-01-18T10:20:26Z

src/main/java/io/kestra/plugin/azure/eventhubs/Consume.java

+
+            Map<EventHubNamePartition, Integer> result = service.poll(
+                consumerContext,
+                new EventHubConsumerService.EventProcessorListener() {


This seems overly complicated to use a listener.
Can't we do without it? After all, the service can return the URI and that it.

Yes, it's perfectly possible to do things differently, but I'm not sure it will be any simpler than it already is. The listener allows to only have to look to the task to understand how events are captured and stored without the need to digging into the consumer polling loop.

src/main/java/io/kestra/plugin/azure/eventhubs/Produce.java

loicmathieu

LGTM, I spotted some missing @PluginProperty please add the before merging.

loicmathieu · 2024-01-19T09:43:15Z

src/main/java/io/kestra/plugin/azure/eventhubs/Produce.java

+@SuperBuilder
+@Getter
+@NoArgsConstructor
+public class Produce extends AbstractEventHubTask implements RunnableTask<Produce.Output> {


It's missing some @PluginProperty

This commit adds the following tasks types: * io.kesta.plugin.azure.messaging.eventhubs.Produce * io.kesta.plugin.azure.messaging.eventhubs.Consume * io.kesta.plugin.azure.messaging.eventhubs.Trigger

fhussonnois changed the title ~~WIP feat(plugins): add new task Produce for EventHubs (#38)~~ WIP feat(plugins): add new plugin tasks for EventHubs (#38) Jan 10, 2024

fhussonnois force-pushed the feature/eventhubs-plugins branch 5 times, most recently from 8d8f63b to 7368497 Compare January 12, 2024 12:40

fhussonnois force-pushed the feature/eventhubs-plugins branch 4 times, most recently from 409db3d to 72540c3 Compare January 12, 2024 14:51

loicmathieu requested changes Jan 12, 2024

View reviewed changes

fhussonnois force-pushed the feature/eventhubs-plugins branch 6 times, most recently from ad0096a to 7668250 Compare January 17, 2024 20:06

loicmathieu requested changes Jan 18, 2024

View reviewed changes

fhussonnois force-pushed the feature/eventhubs-plugins branch 4 times, most recently from a8f5ddc to ca6b7cf Compare January 19, 2024 09:16

loicmathieu approved these changes Jan 19, 2024

View reviewed changes

fhussonnois force-pushed the feature/eventhubs-plugins branch from ca6b7cf to b86c976 Compare January 19, 2024 10:17

feat: add new tasks for Azure EventHubs (kestra-io#38)

9426b80

This commit adds the following tasks types: * io.kesta.plugin.azure.messaging.eventhubs.Produce * io.kesta.plugin.azure.messaging.eventhubs.Consume * io.kesta.plugin.azure.messaging.eventhubs.Trigger

fhussonnois force-pushed the feature/eventhubs-plugins branch from b86c976 to 9426b80 Compare January 26, 2024 14:29

fhussonnois changed the title ~~WIP feat(plugins): add new plugin tasks for EventHubs (#38)~~ feat(plugins): add new plugin tasks for EventHubs (#38) Jan 26, 2024

fhussonnois merged commit 09fb9b6 into kestra-io:master Jan 26, 2024
1 check failed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat(plugins): add new plugin tasks for EventHubs (#38) #56

feat(plugins): add new plugin tasks for EventHubs (#38) #56

fhussonnois commented Jan 10, 2024 •

edited

Loading

anna-geller commented Jan 12, 2024

loicmathieu left a comment

loicmathieu Jan 12, 2024

fhussonnois Jan 12, 2024

loicmathieu left a comment

loicmathieu Jan 18, 2024

fhussonnois Jan 19, 2024

loicmathieu left a comment

loicmathieu Jan 19, 2024

feat(plugins): add new plugin tasks for EventHubs (#38) #56

feat(plugins): add new plugin tasks for EventHubs (#38) #56

Conversation

fhussonnois commented Jan 10, 2024 • edited Loading

What changes are being made and why?

How the changes have been QAed?

Setup Instructions

anna-geller commented Jan 12, 2024

loicmathieu left a comment

Choose a reason for hiding this comment

loicmathieu Jan 12, 2024

Choose a reason for hiding this comment

fhussonnois Jan 12, 2024

Choose a reason for hiding this comment

loicmathieu left a comment

Choose a reason for hiding this comment

loicmathieu Jan 18, 2024

Choose a reason for hiding this comment

fhussonnois Jan 19, 2024

Choose a reason for hiding this comment

loicmathieu left a comment

Choose a reason for hiding this comment

loicmathieu Jan 19, 2024

Choose a reason for hiding this comment

fhussonnois commented Jan 10, 2024 •

edited

Loading