mimaison commented on a change in pull request #9549:
URL: https://github.com/apache/kafka/pull/9549#discussion_r588262888
##########
File path:
connect/runtime/src/main/java/org/apache/kafka/connect/tools/TransformationDoc.java
##########
@@ -18,11 +18,15 @@
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.connect.transforms.Cast;
+import org.apache.kafka.connect.transforms.DropHeaders;
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.HeaderFrom;
+import org.apache.kafka.connect.transforms.HeaderTo;
Review comment:
We don't have this transformation!
##########
File path:
connect/transforms/src/main/java/org/apache/kafka/connect/transforms/DropHeaders.java
##########
@@ -0,0 +1,66 @@
+/*
+ * 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;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.connect.connector.ConnectRecord;
+import org.apache.kafka.connect.header.Headers;
+import org.apache.kafka.connect.transforms.util.SimpleConfig;
+
+import java.util.List;
+import java.util.Map;
+
+public class DropHeaders<R extends ConnectRecord<R>> implements
Transformation<R> {
+
+ public static final String OVERVIEW_DOC =
+ "Removes one or more headers from each record.";
+
+ public static final String HEADERS_FIELD = "headers";
+
+ public static final ConfigDef CONFIG_DEF = new ConfigDef()
+ .define(HEADERS_FIELD, ConfigDef.Type.LIST,
ConfigDef.Importance.HIGH,
+ "The name of the headers to be removed.");
+
+ private List<String> headers;
+
+ @Override
+ public R apply(R record) {
+ Headers updatedHeaders = record.headers().duplicate();
+ for (String name : headers) {
+ updatedHeaders.remove(name);
Review comment:
Because `Headers` is a `LinkedList`, `remove()` has to iterate the whole
list each time. I wonder if we could instead start from an empty headers list
and add the headers not being removed?
##########
File path:
connect/transforms/src/main/java/org/apache/kafka/connect/transforms/InsertHeader.java
##########
@@ -0,0 +1,72 @@
+/*
+ * 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;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.connect.connector.ConnectRecord;
+import org.apache.kafka.connect.data.Schema;
Review comment:
Unused import
##########
File path:
connect/transforms/src/main/java/org/apache/kafka/connect/transforms/DropHeaders.java
##########
@@ -0,0 +1,66 @@
+/*
+ * 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;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.connect.connector.ConnectRecord;
+import org.apache.kafka.connect.header.Headers;
+import org.apache.kafka.connect.transforms.util.SimpleConfig;
+
+import java.util.List;
+import java.util.Map;
+
+public class DropHeaders<R extends ConnectRecord<R>> implements
Transformation<R> {
+
+ public static final String OVERVIEW_DOC =
+ "Removes one or more headers from each record.";
+
+ public static final String HEADERS_FIELD = "headers";
+
+ public static final ConfigDef CONFIG_DEF = new ConfigDef()
+ .define(HEADERS_FIELD, ConfigDef.Type.LIST,
ConfigDef.Importance.HIGH,
+ "The name of the headers to be removed.");
+
+ private List<String> headers;
+
+ @Override
+ public R apply(R record) {
+ Headers updatedHeaders = record.headers().duplicate();
+ for (String name : headers) {
+ updatedHeaders.remove(name);
+ }
+ return record.newRecord(record.topic(), record.kafkaPartition(),
record.keySchema(), record.key(),
+ record.valueSchema(), record.value(), record.timestamp(),
updatedHeaders);
+ }
+
Review comment:
nit, extra line
##########
File path:
connect/transforms/src/main/java/org/apache/kafka/connect/transforms/DropHeaders.java
##########
@@ -0,0 +1,66 @@
+/*
+ * 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;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.connect.connector.ConnectRecord;
+import org.apache.kafka.connect.header.Headers;
+import org.apache.kafka.connect.transforms.util.SimpleConfig;
+
+import java.util.List;
+import java.util.Map;
+
+public class DropHeaders<R extends ConnectRecord<R>> implements
Transformation<R> {
+
+ public static final String OVERVIEW_DOC =
+ "Removes one or more headers from each record.";
+
+ public static final String HEADERS_FIELD = "headers";
+
+ public static final ConfigDef CONFIG_DEF = new ConfigDef()
+ .define(HEADERS_FIELD, ConfigDef.Type.LIST,
ConfigDef.Importance.HIGH,
+ "The name of the headers to be removed.");
+
+ private List<String> headers;
+
+ @Override
+ public R apply(R record) {
+ Headers updatedHeaders = record.headers().duplicate();
+ for (String name : headers) {
+ updatedHeaders.remove(name);
+ }
+ return record.newRecord(record.topic(), record.kafkaPartition(),
record.keySchema(), record.key(),
+ record.valueSchema(), record.value(), record.timestamp(),
updatedHeaders);
+ }
+
+
+ @Override
+ public ConfigDef config() {
+ return CONFIG_DEF;
+ }
+
+ @Override
+ public void close() {
+
Review comment:
nit, extra line
##########
File path:
connect/transforms/src/main/java/org/apache/kafka/connect/transforms/HeaderFrom.java
##########
@@ -0,0 +1,233 @@
+/*
+ * 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;
+
+import org.apache.kafka.common.cache.Cache;
+import org.apache.kafka.common.cache.LRUCache;
+import org.apache.kafka.common.cache.SynchronizedCache;
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.connect.connector.ConnectRecord;
+import org.apache.kafka.connect.data.Field;
+import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.SchemaBuilder;
+import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.header.Header;
+import org.apache.kafka.connect.header.Headers;
+import org.apache.kafka.connect.transforms.util.Requirements;
+import org.apache.kafka.connect.transforms.util.SchemaUtil;
+import org.apache.kafka.connect.transforms.util.SimpleConfig;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static java.lang.String.format;
+import static org.apache.kafka.common.config.ConfigDef.NO_DEFAULT_VALUE;
+
+public abstract class HeaderFrom<R extends ConnectRecord<R>> implements
Transformation<R> {
+
+ public static final String FIELDS_FIELD = "fields";
+ public static final String HEADERS_FIELD = "headers";
+ public static final String OPERATION_FIELD = "operation";
+ private static final String MOVE_OPERATION = "move";
+ private static final String COPY_OPERATION = "copy";
+
+ public static final String OVERVIEW_DOC =
+ "Moves or copies fields in the key/value of a record into that
record's headers. " +
+ "Corresponding elements of <code>" + FIELDS_FIELD +
"</code> and " +
+ "<code>" + HEADERS_FIELD + "</code> together identify a
field and the header it should be " +
+ "moved or copied to. " +
+ "Use the concrete transformation type designed for the
record " +
+ "key (<code>" + Key.class.getName() + "</code>) or value
(<code>" + Value.class.getName() + "</code>).";
+
+ public static final ConfigDef CONFIG_DEF = new ConfigDef()
+ .define(FIELDS_FIELD, ConfigDef.Type.LIST,
ConfigDef.Importance.HIGH,
+ "Field names in the record whose values are to be copied
or moved to headers.")
+ .define(HEADERS_FIELD, ConfigDef.Type.LIST,
ConfigDef.Importance.HIGH,
+ "Header names, in the same order as the field names listed
in the fields configuration property.")
+ .define(OPERATION_FIELD, ConfigDef.Type.STRING, NO_DEFAULT_VALUE,
+ ConfigDef.ValidString.in("move", "copy"),
ConfigDef.Importance.HIGH,
Review comment:
Should we reuse `MOVE_OPERATION` and `COPY_OPERATION` here?
##########
File path:
connect/transforms/src/main/java/org/apache/kafka/connect/transforms/InsertHeader.java
##########
@@ -0,0 +1,72 @@
+/*
+ * 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;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.apache.kafka.connect.connector.ConnectRecord;
+import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.data.SchemaAndValue;
+import org.apache.kafka.connect.data.Values;
+import org.apache.kafka.connect.header.Headers;
+import org.apache.kafka.connect.transforms.util.SimpleConfig;
+
+import java.util.Map;
+
+public class InsertHeader<R extends ConnectRecord<R>> implements
Transformation<R> {
+
+ public static final String OVERVIEW_DOC =
+ "Add a header to each record.";
+
+ public static final String HEADER_FIELD = "header";
+ public static final String VALUE_LITERAL_FIELD = "value.literal";
+
+ public static final ConfigDef CONFIG_DEF = new ConfigDef()
+ .define(HEADER_FIELD, ConfigDef.Type.STRING,
ConfigDef.Importance.HIGH,
+ "The name of the header.")
+ .define(VALUE_LITERAL_FIELD, ConfigDef.Type.STRING,
ConfigDef.Importance.HIGH,
Review comment:
Should we enforce these fields are not null?
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
