[improve][io] support kafka connect transforms and predicates

[efcasado]

Motivation

The current implementation of the Kafka Connect adaptor in Pulsar IO does not support Kafka Connect transforms and predicates. This limits the flexibility and potential of using Kafka Connect-based source connectors, such as the Debezium PostgreSQL source connector, which rely heavily on transformations to modify, filter, or enrich records before they are ingested. By adding support for transforms, we enable users to fully leverage these connectors' capabilities, sparing them the need to deploy additional Pulsar Functions to perform simple message-level transformations.

Modifications

Extended KafkaConnectSource with three additional steps: initPredicates, initTransforms and applyTransforms.

• initPredicates and initTransforms are called during the initialization of the Kafka Connect adaptor and are responsible for setting up any configured predicates and transforms.

• applyTransforms is executed during processRecords and applies all available transforms to each source record before it is processed.

kafka-connect-adaptor has been updated to include the connect-transforms package.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Extended the existing KafkaConnectSourceTest test to verify that transforms can be initialized and applied

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

[github-actions]

@efcasado Please add the following content to your PR description and select a checkbox:

- [ ] `doc` <!-- Your PR contains doc changes -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->

[lhotari]

Thanks for the contribution, @efcasado!

[lhotari]

LGTM

[lhotari]

Connectors that rely on transforms, such as Debezium, typically package their own set of Kafka Connect transforms, making it unnecessary to include them in the adaptor itself.

@efcasado Are you also planning to contribute to the Pulsar IO Debezium connectors?

Debezium version is outdated in Pulsar:

pulsar/pom.xml

Line 204 in a4a3409

<debezium.version>1.9.7.Final</debezium.version>

It would be great to get that updated too. Would you be interested in handling that too? Previous upgrading attempt was #23078, but that's stalled.
Could we get Debezium upgraded to 3.1.x ?

[lhotari]

@efcasado In SMT, there's also predicates (Confluent doc for predicates). Do you have intention to add support for that too?

[efcasado]

@efcasado In SMT, there's also predicates (Confluent doc for predicates). Do you have intention to add support for that too?

I guess it makes sense to include them in this PR since they are somewhat related. I can give it a try unless you tell me to only do this in a separate PR.

[efcasado]

Connectors that rely on transforms, such as Debezium, typically package their own set of Kafka Connect transforms, making it unnecessary to include them in the adaptor itself.

@efcasado Are you also planning to contribute to the Pulsar IO Debezium connectors?

Debezium version is outdated in Pulsar:

pulsar/pom.xml

Line 204 in a4a3409

<debezium.version>1.9.7.Final</debezium.version>

It would be great to get that updated too. Would you be interested in handling that too? Previous upgrading attempt was #23078, but that's stalled.
Could we get Debezium upgraded to 3.1.x ?

Actually, that's how this PR started but I ended up opting for the path of least resistance 😅 I got a version of Debezium 3.1 partially working, but I ran into some issues I could not fix. I will be very happy to give it another try in the coming days.

[lhotari]

@efcasado In SMT, there's also predicates (Confluent doc for predicates). Do you have intention to add support for that too?

I guess it makes sense to include them in this PR since they are somewhat related. I can give it a try unless you tell me to only do this in a separate PR.

It's fine to do it in this PR.

[lhotari]

Good work! Added a comment about supporting negate for predicates. It would be great to have some unit test coverage too.

[efcasado]

@lhotari I believe I addressed all the above points and the change is ready for another round of reviews 😊

[lhotari]

@lhotari I believe I addressed all the above points and the change is ready for another round of reviews 😊

@efcasado great work! one more question about skipping source records.

[lhotari]

LGTM

[codecov-commenter]

Codecov Report

Attention: Patch coverage is 81.01266% with 15 lines in your changes missing coverage. Please review.

Project coverage is 74.18%. Comparing base (bbc6224) to head (9b7134a).
Report is 1057 commits behind head on master.

Files with missing lines	Patch %	Lines
...he/pulsar/io/kafka/connect/KafkaConnectSource.java	82.05%	9 Missing and 5 partials ⚠️
...r/io/kafka/connect/AbstractKafkaConnectSource.java	0.00%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #24221      +/-   ##
============================================
+ Coverage     73.57%   74.18%   +0.61%     
+ Complexity    32624    32555      -69     
============================================
  Files          1877     1866      -11     
  Lines        139502   144919    +5417     
  Branches      15299    16557    +1258     
============================================
+ Hits         102638   107508    +4870     
+ Misses        28908    28897      -11     
- Partials       7956     8514     +558     

Flag	Coverage Δ
inttests	26.71% <ø> (+2.12%)	⬆️
systests	23.28% <ø> (-1.04%)	⬇️
unittests	73.66% <81.01%> (+0.82%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...r/io/kafka/connect/AbstractKafkaConnectSource.java	63.28% <0.00%> (+1.68%)	⬆️
...he/pulsar/io/kafka/connect/KafkaConnectSource.java	68.00% <82.05%> (+13.94%)	⬆️

... and 1078 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[lhotari]

Great work on this PR @efcasado! Merging this soon.

Actually, that's how this PR started but I ended up opting for the path of least resistance 😅 I got a version of Debezium 3.1 partially working, but I ran into some issues I could not fix. I will be very happy to give it another try in the coming days.

Looking forward to future contributions!

[efcasado]

@lhotari Will this change be cherry-picked for 3.x too? We are running Pulsar 3.3.5 and we are wondering if we should just use our own fork or wait for you and the team to promote this change to the 3.x branch.

[lhotari]

@lhotari Will this change be cherry-picked for 3.x too? We are running Pulsar 3.3.5 and we are wondering if we should just use our own fork or wait for you and the team to promote this change to the 3.x branch.

I don't see barriers in cherry-picking to 3.3.x since this isn't introducing breaking changes. I'll mark it for 3.0.x and 3.3.x too.

[lhotari]

@efcasado FYI, There's an open issue #24646 related to this PR. Being fixed in #24654