Add record is null smt

src/main/java/io/lenses/connect/smt/header/RecordIsNull.java

@@ -0,0 +1,51 @@
+/**
+ * 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 io.lenses.connect.smt.header;
+
+import java.util.Map;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.utils.AppInfoParser;
+import org.apache.kafka.connect.components.Versioned;
+import org.apache.kafka.connect.connector.ConnectRecord;
+import org.apache.kafka.connect.transforms.predicates.Predicate;
+
+public class RecordIsNull<R extends ConnectRecord<R>> implements Predicate<R>, Versioned {
+
+  public static final String OVERVIEW_DOC =
+      "A predicate which is true for records which are tombstones (i.e. have null value).";
+  public static final ConfigDef CONFIG_DEF = new ConfigDef();
+
+  @Override
+  public String version() {
+    return AppInfoParser.getVersion();
+  }
+
+  @Override
+  public ConfigDef config() {
+    return CONFIG_DEF;
+  }
+
+  @Override
+  public boolean test(R record) {
+    return record == null;
+  }
+
+  @Override
+  public void close() {}
+
+  @Override
+  public void configure(Map<String, ?> configs) {}
+
+  @Override
+  public String toString() {
+    return "RecordIsTombstone{}";
+  }
+}