Merge pull request #8 from aiven/tombstone-handler

ivanyu · web-flow · commit 9e819a71ae23 · 2020-03-09T12:08:52.000+02:00
Add TombstoneHandler
diff --git a/README.md b/README.md
@@ -46,14 +46,32 @@ The transformation defines the following configurations:
 - `field.name` - The name of the field which should be used as the topic name. If `null` or empty, the entire key or value is used (and assumed to be a string). By default is `null`.
 - `skip.missing.or.null` - In case the source of the new topic name is `null` or missing, should a record be silently passed without transformation. By default is `false`.
 
-Here's an example of this transformation configuration:
+Here is an example of this transformation configuration:
 
 ```properties
 transforms=ExtractTopicFromValueField
 transforms.ExtractTopicFromValueField.type=io.aiven.kafka.connect.transforms.ExtractTopic$Value
 transforms.ExtractTopicFromValueField.field.name=inner_field_name
 ```
 
+### `TombstoneHandler`
+
+This transformation manages tombstone records, 
+i.e. records with the entire value field being `null`.
+
+The transformation defines the following configurations:
+- `behavior` - The action the transformation must perform when encounter a tombstone record. The supported values are:
+  - `drop_silent` - silently drop tombstone records.
+  - `drop_warn` - drop tombstone records and log at `WARN` level.
+  - `fail` - fail with `DataException`.
+
+Here is an example of this transformation configuration:
+```properties
+transforms=TombstoneHandler
+transforms.TombstoneHandler.type=io.aiven.kafka.connect.transforms.TombstoneHandler
+transforms.TombstoneHandler.behavior=drop_silent
+```
+
 ## License
 
 This project is licensed under the [Apache License, Version 2.0](LICENSE).
diff --git a/config/checkstyle/java.header b/config/checkstyle/java.header
@@ -1,5 +1,5 @@
 /\*
- \* Copyright 2019 Aiven Oy
+ \* Copyright 20(19|2[0-9]) Aiven Oy
  \*
  \* Licensed under the Apache License, Version 2.0 \(the "License"\);
  \* you may not use this file except in compliance with the License.
diff --git a/src/integration-test/java/io/aiven/kafka/connect/transforms/TestSourceConnector.java b/src/integration-test/java/io/aiven/kafka/connect/transforms/TestSourceConnector.java
@@ -53,7 +53,7 @@ public Class<? extends Task> taskClass() {
 
     @Override
     public List<Map<String, String>> taskConfigs(final int maxTasks) {
-        return Collections.singletonList(Collections.EMPTY_MAP);
+        return Collections.singletonList(Collections.emptyMap());
     }
 
     @Override
diff --git a/src/main/java/io/aiven/kafka/connect/transforms/TombstoneHandler.java b/src/main/java/io/aiven/kafka/connect/transforms/TombstoneHandler.java
@@ -0,0 +1,88 @@
+/*
+ * Copyright 2020 Aiven Oy
+ *
+ * Licensed 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.aiven.kafka.connect.transforms;
+
+import java.util.Map;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.connect.connector.ConnectRecord;
+import org.apache.kafka.connect.errors.DataException;
+import org.apache.kafka.connect.transforms.Transformation;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class TombstoneHandler<R extends ConnectRecord<R>> implements Transformation<R> {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(TombstoneHandler.class);
+
+    private TombstoneHandlerConfig tombstoneHandlerConfig;
+
+    @Override
+    public ConfigDef config() {
+        return TombstoneHandlerConfig.config();
+    }
+
+    @Override
+    public void configure(final Map<String, ?> configs) {
+        this.tombstoneHandlerConfig = new TombstoneHandlerConfig(configs);
+    }
+
+    @Override
+    public R apply(final R record) {
+        if (record.value() == null) {
+            behaveOnNull(record);
+            return null;
+        } else {
+            return record;
+        }
+    }
+
+    private void behaveOnNull(final ConnectRecord<R> r) {
+
+        switch (tombstoneHandlerConfig.getBehavior()) {
+            case FAIL:
+                throw new DataException(
+                    String.format(
+                        "Tombstone record encountered, failing due to configured '%s' behavior",
+                        TombstoneHandlerConfig.Behavior.FAIL.name().toLowerCase()
+                    )
+                );
+            case DROP_SILENT:
+                LOGGER.debug(
+                    "Tombstone record encountered, dropping due to configured '{}' behavior",
+                    TombstoneHandlerConfig.Behavior.DROP_SILENT.name().toLowerCase()
+                );
+                break;
+            case DROP_WARN:
+                LOGGER.debug(
+                    "Tombstone record encountered, dropping due to configured '{}' behavior",
+                    TombstoneHandlerConfig.Behavior.DROP_WARN.name().toLowerCase()
+                );
+                break;
+            default:
+                throw new DataException(
+                    String.format("Unknown behavior: %s", tombstoneHandlerConfig.getBehavior())
+                );
+        }
+    }
+
+    @Override
+    public void close() {
+    }
+
+}
diff --git a/src/main/java/io/aiven/kafka/connect/transforms/TombstoneHandlerConfig.java b/src/main/java/io/aiven/kafka/connect/transforms/TombstoneHandlerConfig.java
@@ -0,0 +1,107 @@
+/*
+ * Copyright 2020 Aiven Oy
+ *
+ * Licensed 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.aiven.kafka.connect.transforms;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+
+public final class TombstoneHandlerConfig extends AbstractConfig {
+
+    public static final String TOMBSTONE_HANDLER_BEHAVIOR = "behavior";
+
+    public TombstoneHandlerConfig(final Map<?, ?> originals) {
+        super(config(), originals);
+    }
+
+    static ConfigDef config() {
+        return new ConfigDef()
+            .define(
+                TOMBSTONE_HANDLER_BEHAVIOR,
+                ConfigDef.Type.STRING,
+                ConfigDef.NO_DEFAULT_VALUE,
+                new ConfigDef.Validator() {
+                    @Override
+                    public void ensureValid(final String name, final Object value) {
+                        assert value instanceof String;
+
+                        final String strValue = (String) value;
+
+                        if (Objects.isNull(strValue) || strValue.isEmpty()) {
+                            throw new ConfigException(
+                                TOMBSTONE_HANDLER_BEHAVIOR,
+                                value,
+                                "String must be non-empty");
+                        }
+
+                        try {
+                            Behavior.of(strValue);
+                        } catch (final IllegalArgumentException e) {
+                            throw new ConfigException(
+                                TOMBSTONE_HANDLER_BEHAVIOR,
+                                value,
+                                e.getMessage());
+                        }
+                    }
+                },
+                ConfigDef.Importance.MEDIUM,
+                String.format(
+                    "Specifies the behavior on encountering tombstone messages. Possible values are: %s",
+                    Behavior.BEHAVIOR_NAMES
+                )
+            );
+    }
+
+    public Behavior getBehavior() {
+        return Behavior.of(getString(TOMBSTONE_HANDLER_BEHAVIOR));
+    }
+
+    public enum Behavior {
+
+        DROP_SILENT,
+        DROP_WARN,
+        FAIL;
+
+        private static final List<String> BEHAVIOR_NAMES =
+            Arrays.stream(Behavior.values())
+                .map(b -> b.name().toLowerCase())
+                .collect(Collectors.toList());
+
+        public static Behavior of(final String v) {
+
+            for (final Behavior b : Behavior.values()) {
+                if (b.name().equalsIgnoreCase(v)) {
+                    return b;
+                }
+            }
+            throw new IllegalArgumentException(
+                String.format(
+                    "Unsupported behavior name: %s. Supported are: %s", v,
+                        String.join(",", BEHAVIOR_NAMES))
+            );
+
+        }
+
+    }
+
+}
diff --git a/src/test/java/io/aiven/kafka/connect/transforms/TombstoneHandlerConfigTest.java b/src/test/java/io/aiven/kafka/connect/transforms/TombstoneHandlerConfigTest.java
@@ -0,0 +1,86 @@
+/*
+ * Copyright 2020 Aiven Oy
+ *
+ * Licensed 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.aiven.kafka.connect.transforms;
+
+import java.util.Map;
+
+import org.apache.kafka.common.config.ConfigException;
+
+import org.junit.jupiter.api.Test;
+import org.testcontainers.shaded.com.google.common.collect.ImmutableMap;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+final class TombstoneHandlerConfigTest {
+
+    @Test
+    final void failOnUnknownBehaviorName() {
+        final Throwable t =
+            assertThrows(
+                ConfigException.class,
+                () -> new TombstoneHandlerConfig(newBehaviorProps("asdasdsadas"))
+            );
+        assertEquals(
+            "Invalid value asdasdsadas for configuration behavior: "
+                + "Unsupported behavior name: asdasdsadas. Supported are: drop_silent,drop_warn,fail",
+            t.getMessage()
+        );
+    }
+
+    @Test
+    final void acceptCorrectBehaviorNames() {
+
+        TombstoneHandlerConfig c =
+            new TombstoneHandlerConfig(
+                newBehaviorProps(
+                    TombstoneHandlerConfig.Behavior.DROP_SILENT.name()
+                )
+            );
+        assertEquals(TombstoneHandlerConfig.Behavior.DROP_SILENT, c.getBehavior());
+
+        c =
+            new TombstoneHandlerConfig(
+                newBehaviorProps(
+                    TombstoneHandlerConfig.Behavior.FAIL.name().toLowerCase()
+                )
+            );
+        assertEquals(TombstoneHandlerConfig.Behavior.FAIL, c.getBehavior());
+
+        c =
+            new TombstoneHandlerConfig(
+                newBehaviorProps(
+                    "Drop_WArn"
+                )
+            );
+        assertEquals(TombstoneHandlerConfig.Behavior.DROP_WARN, c.getBehavior());
+    }
+
+    @Test
+    final void failOnEmptyBehaviorName() {
+        final Throwable t = assertThrows(
+            ConfigException.class,
+            () -> new TombstoneHandlerConfig(newBehaviorProps(""))
+        );
+        assertEquals("Invalid value  for configuration behavior: String must be non-empty", t.getMessage());
+    }
+
+    private Map<String, String> newBehaviorProps(final String bv) {
+        return ImmutableMap.of(TombstoneHandlerConfig.TOMBSTONE_HANDLER_BEHAVIOR, bv);
+    }
+
+}
diff --git a/src/test/java/io/aiven/kafka/connect/transforms/TombstoneHandlerTest.java b/src/test/java/io/aiven/kafka/connect/transforms/TombstoneHandlerTest.java

Original file line number	Diff line number	Diff line change
`@@ -53,7 +53,7 @@ public Class<? extends Task> taskClass() {`
`53`	`53`
`54`	`54`	`@Override`
`55`	`55`	`public List<Map<String, String>> taskConfigs(final int maxTasks) {`
`56`		`- return Collections.singletonList(Collections.EMPTY_MAP);`
	`56`	`+ return Collections.singletonList(Collections.emptyMap());`
`57`	`57`	`}`
`58`	`58`
`59`	`59`	`@Override`