efcasado commented on code in PR #24221:
URL: https://github.com/apache/pulsar/pull/24221#discussion_r2064151095
##########
pulsar-io/kafka-connect-adaptor/src/main/java/org/apache/pulsar/io/kafka/connect/KafkaConnectSource.java:
##########
@@ -95,8 +134,17 @@ private void initTransforms(Map<String, Object> config) {
java.util.Map.Entry::getValue
));
log.info("transform config: {}", transformConfig);
+ String predicateName = (String)
transformConfig.get("predicate");
+ Predicate<SourceRecord> predicate = null;
+ if (predicateName != null) {
+ predicate = predicates.get(predicateName);
+ if (predicate == null) {
+ log.warn("Transform {} references non-existent
predicate: {}",
+ transformName, predicateName);
+ }
+ }
transform.configure(transformConfig);
- transformations.add(transform);
+ transformations.add(Pair.of(predicate, transform));
Review Comment:
Thanks for holding hands with me. As an Elixir developer who hasn't touched
Java in years, I genuinely appreciate your help here :pray: I should have
something ready soon, and I will make sure I add a couple of unit tests to
cover predicates and negated predicates, as suggested.
--
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-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
