You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by "jakubmalek (via GitHub)" <gi...@apache.org> on 2024/08/09 11:45:39 UTC
[I] Azure EventHubs offset checkpoint store [camel-kafka-connector]
jakubmalek opened a new issue, #1662:
URL: https://github.com/apache/camel-kafka-connector/issues/1662
The [camel-azure-eventhubs-source-kafka-connector](https://camel.apache.org/camel-kafka-connector/4.0.x/reference/connectors/camel-azure-eventhubs-source-kafka-source-connector.html) currently requires configuration of the Azure Blob Storage, used for storing the checkpoints.
I think there should be an option to store checkpoints in the Kafka Connect offset topic. The source connector state is usually stored there by including the source offset into the `SourceRecord` with the given source partition used as key for lookup.
The checkpoints can then be loaded on the task start using the `OffsetStorageReader` without need for external storage.
As far as I understand, the related Camel Component allows to set the a custom `CheckpointStore` bean instance to be used by the Azure SDK client. So it should be possible to provide the `CheckpointStore` implementation based on Kafka Connect API and use it instead of the `BlobCheckpointStore`.
I was thinking of doing so with a PR but I'm not sure how to deal with the Kamelet, which currently requires that all parameters related to the Blob Storage are set.
In general I see two possible options:
**Option A. Make Azure Blob parameters in the Kamelet optional**
In this option, the `CamelAzureeventhubssourceSourceTask` can be adopted to initialize `CheckpointStore` based on the offsets if the Blob storage is not configured.
The `CamelSourceTask` could be extended to provide option to customize the created `Endpoint` so the initialized `CheckpointStore` by the task on start can be set via `EventHubsEndpoint` e.g.
`eventHubsEndpoint.getConfiguration().setCheckpointStore(checkpointStore)`
**Option B. Add new Kamelet variant for Azure EventHubs source without parameters for Blob storage**
This options works similarly to Option A, just as a dedicated connector and dedicated Kamelet descriptor.
What do you think about the proposed solutions ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Azure EventHubs offset checkpoint store [camel-kafka-connector]
Posted by "jakubmalek (via GitHub)" <gi...@apache.org>.
jakubmalek commented on issue #1662:
URL: https://github.com/apache/camel-kafka-connector/issues/1662#issuecomment-2278099954
The problem is that `CheckpointStore` requires `OffsetStoreReader` that can be obtained only during task startup.
As I understand I can define a bean with something like this:
`#class:org.apache.camel.kafkaconnector.azureeventhubssource.checkpoint.EventHubOffsetCheckpointStore`
But I'm not sure how can I initialize it, in `CamelSourceTask` with `SourceTaskContext` during task startup.
Another problem is how to apply `sourcePartition` and `sourceOffset` to created `SourceRecord` which is needed to store the checkpoints in the topic.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Azure EventHubs offset checkpoint store [camel-kafka-connector]
Posted by "jakubmalek (via GitHub)" <gi...@apache.org>.
jakubmalek commented on issue #1662:
URL: https://github.com/apache/camel-kafka-connector/issues/1662#issuecomment-2284032554
OK I think I know how this can be implemented now.
1. The checkpoint configuration could be provided as this `#class:org.apache.camel.kafkaconnector.azureeventhubssource.checkpoint.EventHubOffsetCheckpointStore` string so it's initialized by Camel
2. The `EventHubOffsetCheckpointStore` would implement `CamelContextAware` interface, so it can access `CamelContext`
3. The `CamelSourceTask` would bind a bean providing access to `SourceTaskContext` and option to configure provider of the source partition and source offset from the `Exchange`
I still need some time to put it all together, but I think I got the rough idea how it can be setup without need for modification of the generated code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Azure EventHubs offset checkpoint store [camel-kafka-connector]
Posted by "jakubmalek (via GitHub)" <gi...@apache.org>.
jakubmalek commented on issue #1662:
URL: https://github.com/apache/camel-kafka-connector/issues/1662#issuecomment-2277923413
To get the rough understanding what I have in mind when it comes to Option A, you check this [commit](https://github.com/apache/camel-kafka-connector/commit/4b5bfaa018a9cd65e43c9aa06b265fa26761b753#diff-e2091f434ada52b40b90eddb61f035a05a17366f3e539e3479467dbd59cafa24).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [I] Azure EventHubs offset checkpoint store [camel-kafka-connector]
Posted by "oscerd (via GitHub)" <gi...@apache.org>.
oscerd commented on issue #1662:
URL: https://github.com/apache/camel-kafka-connector/issues/1662#issuecomment-2277934359
If you override the source task it will be override once we Regen the connector from the kamelets released catalog. My suggestion would be to mark the parameters as optional and not required, but this would require defining a bean with the checkpoint store. Otherwise the workflow will fail. Create another kamelet is a bit overkilling for this.
Thanks
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org