KAFKA-9673: Filter and Conditional SMTs

clients/src/main/java/org/apache/kafka/common/utils/Utils.java

@@ -342,6 +342,18 @@ public static <T> Class<? extends T> loadClass(String klass, Class<T> base) thro
         return Class.forName(klass, true, Utils.getContextOrKafkaClassLoader()).asSubclass(base);
     }
 
+    /**
+     * Cast {@code klass} to {@code base} and instantiate it.
+     * @param klass The class to instantiate
+     * @param base A know baseclass of klass.
+     * @param <T> the type of the base class
+     * @throws ClassCastException If {@code klass} is not a subclass of {@code base}.
+     * @return the new instance.
+     */
+    public static <T> T newInstance(Class<?> klass, Class<T> base) {
+        return Utils.newInstance(klass.asSubclass(base));
+    }
+
     /**
      * Construct a new object using a class name and parameters.
      *

connect/api/src/main/java/org/apache/kafka/connect/transforms/predicates/Predicate.java

@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.connect.transforms.predicates;
+
+import org.apache.kafka.common.Configurable;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.connect.connector.ConnectRecord;
+
+/**
+ * <p>A predicate on records.
+ * Predicates can be used to conditionally apply a {@link org.apache.kafka.connect.transforms.Transformation}
+ * by configuring the transformation's {@code predicate} (and {@code negate}) configuration parameters.
+ * In particular, the {@code Filter} transformation can be conditionally applied in order to filter
+ * certain records from further processing.
+ *
+ * <p>Implementations of this interface must be public and have a public constructor with no parameters.
+ *
+ * @param <R> The type of record.
+ */
+public interface Predicate<R extends ConnectRecord<R>> extends Configurable, AutoCloseable {
+
+    /**
+     * Configuration specification for this predicate.
+     *
+     * @return the configuration definition for this predicate; never null
+     */
+    ConfigDef config();
+
+    /**
+     * Returns whether the given record satisfies this predicate.
+     *
+     * @param record the record to evaluate; may not be null
+     * @return true if the predicate matches, or false otherwise
+     */
+    boolean test(R record);
+
+    @Override
+    void close();
+}

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/PredicatedTransformation.java

@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.connect.runtime;
+
+import java.util.Map;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.connect.connector.ConnectRecord;
+import org.apache.kafka.connect.errors.ConnectException;
+import org.apache.kafka.connect.transforms.Transformation;
+import org.apache.kafka.connect.transforms.predicates.Predicate;
+
+/**
+ * Decorator for a {@link Transformation} which applies the delegate only when a
+ * {@link Predicate} is true (or false, according to {@code negate}).
+ * @param <R>
+ */
+class PredicatedTransformation<R extends ConnectRecord<R>> implements Transformation<R> {
+
+    static final String PREDICATE_CONFIG = "predicate";
+    static final String NEGATE_CONFIG = "negate";
+    Predicate<R> predicate;
+    Transformation<R> delegate;
+    boolean negate;
+
+    PredicatedTransformation(Predicate<R> predicate, boolean negate, Transformation<R> delegate) {
+        this.predicate = predicate;
+        this.negate = negate;
+        this.delegate = delegate;
+    }
+
+    @Override
+    public void configure(Map<String, ?> configs) {
+        throw new ConnectException(PredicatedTransformation.class.getName() + ".configure() " +
+                "should never be called directly.");
+    }
+
+    @Override
+    public R apply(R record) {
+        if (negate ^ predicate.test(record)) {
+            return delegate.apply(record);
+        }
+        return record;
+    }
+
+    @Override
+    public ConfigDef config() {
+        throw new ConnectException(PredicatedTransformation.class.getName() + ".config() " +
+                "should never be called directly.");
+    }
+
+    @Override
+    public void close() {
+        Utils.closeQuietly(delegate, "predicated");
+        Utils.closeQuietly(predicate, "predicate");
+    }
+
+    @Override
+    public String toString() {
+        return "PredicatedTransformation{" +
+                "predicate=" + predicate +
+                ", delegate=" + delegate +
+                ", negate=" + negate +
+                '}';
+    }
+}

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/DelegatingClassLoader.java

@@ -24,6 +24,7 @@
 import org.apache.kafka.connect.storage.Converter;
 import org.apache.kafka.connect.storage.HeaderConverter;
 import org.apache.kafka.connect.transforms.Transformation;
+import org.apache.kafka.connect.transforms.predicates.Predicate;
 import org.reflections.Configuration;
 import org.reflections.Reflections;
 import org.reflections.ReflectionsException;
@@ -72,6 +73,7 @@ public class DelegatingClassLoader extends URLClassLoader {
     private final SortedSet<PluginDesc<Converter>> converters;
     private final SortedSet<PluginDesc<HeaderConverter>> headerConverters;
     private final SortedSet<PluginDesc<Transformation>> transformations;
+    private final SortedSet<PluginDesc<Predicate>> predicates;
     private final SortedSet<PluginDesc<ConfigProvider>> configProviders;
     private final SortedSet<PluginDesc<ConnectRestExtension>> restExtensions;
     private final SortedSet<PluginDesc<ConnectorClientConfigOverridePolicy>> connectorClientConfigPolicies;
@@ -92,6 +94,7 @@ public DelegatingClassLoader(List<String> pluginPaths, ClassLoader parent) {
         this.converters = new TreeSet<>();
         this.headerConverters = new TreeSet<>();
         this.transformations = new TreeSet<>();
+        this.predicates = new TreeSet<>();
         this.configProviders = new TreeSet<>();
         this.restExtensions = new TreeSet<>();
         this.connectorClientConfigPolicies = new TreeSet<>();
@@ -121,6 +124,10 @@ public Set<PluginDesc<Transformation>> transformations() {
         return transformations;
     }
 
+    public Set<PluginDesc<Predicate>> predicates() {
+        return predicates;
+    }
+
     public Set<PluginDesc<ConfigProvider>> configProviders() {
         return configProviders;
     }
@@ -269,6 +276,8 @@ private void scanUrlsAndAddPlugins(
             headerConverters.addAll(plugins.headerConverters());
             addPlugins(plugins.transformations(), loader);
             transformations.addAll(plugins.transformations());
+            addPlugins(plugins.predicates(), loader);
+            predicates.addAll(plugins.predicates());
             addPlugins(plugins.configProviders(), loader);
             configProviders.addAll(plugins.configProviders());
             addPlugins(plugins.restExtensions(), loader);
@@ -329,6 +338,7 @@ private PluginScanResult scanPluginPath(
                 getPluginDesc(reflections, Converter.class, loader),
                 getPluginDesc(reflections, HeaderConverter.class, loader),
                 getPluginDesc(reflections, Transformation.class, loader),
+                getPluginDesc(reflections, Predicate.class, loader),
                 getServiceLoaderPluginDesc(ConfigProvider.class, loader),
                 getServiceLoaderPluginDesc(ConnectRestExtension.class, loader),
                 getServiceLoaderPluginDesc(ConnectorClientConfigOverridePolicy.class, loader)
@@ -402,6 +412,7 @@ private void addAllAliases() {
         addAliases(converters);
         addAliases(headerConverters);
         addAliases(transformations);
+        addAliases(predicates);
         addAliases(restExtensions);
         addAliases(connectorClientConfigPolicies);
     }

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginScanResult.java

@@ -23,6 +23,7 @@
 import org.apache.kafka.connect.storage.Converter;
 import org.apache.kafka.connect.storage.HeaderConverter;
 import org.apache.kafka.connect.transforms.Transformation;
+import org.apache.kafka.connect.transforms.predicates.Predicate;
 
 import java.util.Arrays;
 import java.util.Collection;
@@ -33,6 +34,7 @@ public class PluginScanResult {
     private final Collection<PluginDesc<Converter>> converters;
     private final Collection<PluginDesc<HeaderConverter>> headerConverters;
     private final Collection<PluginDesc<Transformation>> transformations;
+    private final Collection<PluginDesc<Predicate>> predicates;
     private final Collection<PluginDesc<ConfigProvider>> configProviders;
     private final Collection<PluginDesc<ConnectRestExtension>> restExtensions;
     private final Collection<PluginDesc<ConnectorClientConfigOverridePolicy>> connectorClientConfigPolicies;
@@ -44,6 +46,7 @@ public PluginScanResult(
             Collection<PluginDesc<Converter>> converters,
             Collection<PluginDesc<HeaderConverter>> headerConverters,
             Collection<PluginDesc<Transformation>> transformations,
+            Collection<PluginDesc<Predicate>> predicates,
             Collection<PluginDesc<ConfigProvider>> configProviders,
             Collection<PluginDesc<ConnectRestExtension>> restExtensions,
             Collection<PluginDesc<ConnectorClientConfigOverridePolicy>> connectorClientConfigPolicies
@@ -52,6 +55,7 @@ public PluginScanResult(
         this.converters = converters;
         this.headerConverters = headerConverters;
         this.transformations = transformations;
+        this.predicates = predicates;
         this.configProviders = configProviders;
         this.restExtensions = restExtensions;
         this.connectorClientConfigPolicies = connectorClientConfigPolicies;
@@ -76,6 +80,10 @@ public Collection<PluginDesc<Transformation>> transformations() {
         return transformations;
     }
 
+    public Collection<PluginDesc<Predicate>> predicates() {
+        return predicates;
+    }
+
     public Collection<PluginDesc<ConfigProvider>> configProviders() {
         return configProviders;
     }

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/PluginUtils.java

@@ -128,7 +128,7 @@ public class PluginUtils {
     // added to the WHITELIST), then this base interface or class needs to be excluded in the
     // regular expression pattern
     private static final Pattern WHITELIST = Pattern.compile("^org\\.apache\\.kafka\\.(?:connect\\.(?:"
-            + "transforms\\.(?!Transformation$).*"
+            + "transforms\\.(?!Transformation|predicates\\.Predicate$).*"
             + "|json\\..*"
             + "|file\\..*"
             + "|mirror\\..*"

connect/runtime/src/main/java/org/apache/kafka/connect/runtime/isolation/Plugins.java

@@ -33,6 +33,7 @@
 import org.apache.kafka.connect.storage.ConverterType;
 import org.apache.kafka.connect.storage.HeaderConverter;
 import org.apache.kafka.connect.transforms.Transformation;
+import org.apache.kafka.connect.transforms.predicates.Predicate;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -167,6 +168,10 @@ public Set<PluginDesc<Transformation>> transformations() {
         return delegatingLoader.transformations();
     }
 
+    public Set<PluginDesc<Predicate>> predicates() {
+        return delegatingLoader.predicates();
+    }
+
     public Set<PluginDesc<ConfigProvider>> configProviders() {
         return delegatingLoader.configProviders();
     }

connect/runtime/src/main/java/org/apache/kafka/connect/tools/TransformationDoc.java

@@ -19,6 +19,7 @@
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.connect.transforms.Cast;
 import org.apache.kafka.connect.transforms.ExtractField;
+import org.apache.kafka.connect.transforms.Filter;
 import org.apache.kafka.connect.transforms.Flatten;
 import org.apache.kafka.connect.transforms.HoistField;
 import org.apache.kafka.connect.transforms.InsertField;
@@ -60,7 +61,8 @@ private DocInfo(String transformationName, String overview, ConfigDef configDef)
             new DocInfo(RegexRouter.class.getName(), RegexRouter.OVERVIEW_DOC, RegexRouter.CONFIG_DEF),
             new DocInfo(Flatten.class.getName(), Flatten.OVERVIEW_DOC, Flatten.CONFIG_DEF),
             new DocInfo(Cast.class.getName(), Cast.OVERVIEW_DOC, Cast.CONFIG_DEF),
-            new DocInfo(TimestampConverter.class.getName(), TimestampConverter.OVERVIEW_DOC, TimestampConverter.CONFIG_DEF)
+            new DocInfo(TimestampConverter.class.getName(), TimestampConverter.OVERVIEW_DOC, TimestampConverter.CONFIG_DEF),
+            new DocInfo(Filter.class.getName(), Filter.OVERVIEW_DOC, Filter.CONFIG_DEF)
     );
 
     private static void printTransformationHtml(PrintStream out, DocInfo docInfo) {

connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorHandle.java

@@ -17,6 +17,7 @@
 package org.apache.kafka.connect.integration;
 
 import org.apache.kafka.connect.errors.DataException;
+import org.apache.kafka.connect.sink.SinkRecord;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -26,6 +27,7 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -57,7 +59,19 @@ public ConnectorHandle(String connectorName) {
      * @return a non-null {@link TaskHandle}
      */
     public TaskHandle taskHandle(String taskId) {
-        return taskHandles.computeIfAbsent(taskId, k -> new TaskHandle(this, taskId));
+        return taskHandle(taskId, null);
+    }
+
+    /**
+     * Get or create a task handle for a given task id. The task need not be created when this method is called. If the
+     * handle is called before the task is created, the task will bind to the handle once it starts (or restarts).
+     *
+     * @param taskId the task id
+     * @param consumer A callback invoked when a sink task processes a record.
+     * @return a non-null {@link TaskHandle}
+     */
+    public TaskHandle taskHandle(String taskId, Consumer<SinkRecord> consumer) {
+        return taskHandles.computeIfAbsent(taskId, k -> new TaskHandle(this, taskId, consumer));
     }
 
     /**

connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSinkConnector.java

@@ -125,7 +125,7 @@ public void open(Collection<TopicPartition> partitions) {
         @Override
         public void put(Collection<SinkRecord> records) {
             for (SinkRecord rec : records) {
-                taskHandle.record();
+                taskHandle.record(rec);
                 TopicPartition tp = cachedTopicPartitions
                         .computeIfAbsent(rec.topic(), v -> new HashMap<>())
                         .computeIfAbsent(rec.kafkaPartition(), v -> new TopicPartition(rec.topic(), rec.kafkaPartition()));

connect/runtime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java

@@ -20,6 +20,7 @@
 import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.connect.connector.Task;
 import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.header.ConnectHeaders;
 import org.apache.kafka.connect.runtime.TestSourceConnector;
 import org.apache.kafka.connect.source.SourceRecord;
 import org.apache.kafka.connect.source.SourceTask;
@@ -138,7 +139,9 @@ public List<SourceRecord> poll() {
                                 Schema.STRING_SCHEMA,
                                 "key-" + taskId + "-" + seqno,
                                 Schema.STRING_SCHEMA,
-                                "value-" + taskId + "-" + seqno))
+                                "value-" + taskId + "-" + seqno,
+                                null,
+                                new ConnectHeaders().addLong("header-" + seqno, seqno)))
                         .collect(Collectors.toList());
             }
             return null;