gharris1727 commented on code in PR #13184:
URL: https://github.com/apache/kafka/pull/13184#discussion_r1119389854
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/ConnectorConfig.java:
##########
@@ -268,30 +268,30 @@ public boolean includeRecordDetailsInErrorLog() {
}
/**
- * Returns the initialized list of {@link Transformation} which are
specified in {@link #TRANSFORMS_CONFIG}.
+ * Returns the initialized list of {@link TransformationStage} which are
specified in {@link #TRANSFORMS_CONFIG}.
*/
- public <R extends ConnectRecord<R>> List<Transformation<R>>
transformations() {
+ public <R extends ConnectRecord<R>> List<TransformationStage<R>>
transformationStages() {
final List<String> transformAliases = getList(TRANSFORMS_CONFIG);
- final List<Transformation<R>> transformations = new
ArrayList<>(transformAliases.size());
+ final List<TransformationStage<R>> transformations = new
ArrayList<>(transformAliases.size());
for (String alias : transformAliases) {
final String prefix = TRANSFORMS_CONFIG + "." + alias + ".";
try {
@SuppressWarnings("unchecked")
final Transformation<R> transformation =
Utils.newInstance(getClass(prefix + "type"), Transformation.class);
Map<String, Object> configs = originalsWithPrefix(prefix);
- Object predicateAlias =
configs.remove(PredicatedTransformation.PREDICATE_CONFIG);
- Object negate =
configs.remove(PredicatedTransformation.NEGATE_CONFIG);
+ Object predicateAlias =
configs.remove(TransformationStage.PREDICATE_CONFIG);
+ Object negate =
configs.remove(TransformationStage.NEGATE_CONFIG);
transformation.configure(configs);
if (predicateAlias != null) {
String predicatePrefix = PREDICATES_PREFIX +
predicateAlias + ".";
@SuppressWarnings("unchecked")
Predicate<R> predicate =
Utils.newInstance(getClass(predicatePrefix + "type"), Predicate.class);
predicate.configure(originalsWithPrefix(predicatePrefix));
- transformations.add(new
PredicatedTransformation<>(predicate, negate == null ? false :
Boolean.parseBoolean(negate.toString()), transformation));
+ transformations.add(new TransformationStage<>(predicate,
negate != null && Boolean.parseBoolean(negate.toString()), transformation));
Review Comment:
since there's a disagreement over this among reviewers, i'll leave this as
it was on trunk.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
