forked from jcustenborder/kafka-connect-transform-common
-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Added transformation to filter records based on items in fields. Fixes …
- Loading branch information
1 parent
189cb8a
commit 0f21d20
Showing
4 changed files
with
280 additions
and
29 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
131 changes: 131 additions & 0 deletions
131
src/main/java/com/github/jcustenborder/kafka/connect/transform/common/PatternFilter.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,131 @@ | ||
| /** | ||
| * Copyright © 2017 Jeremy Custenborder ([email protected]) | ||
| * | ||
| * 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 com.github.jcustenborder.kafka.connect.transform.common; | ||
|
|
||
| import com.github.jcustenborder.kafka.connect.utils.config.Description; | ||
| import com.github.jcustenborder.kafka.connect.utils.config.DocumentationTip; | ||
| import com.github.jcustenborder.kafka.connect.utils.config.Title; | ||
| import org.apache.kafka.common.config.ConfigDef; | ||
| 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.SchemaAndValue; | ||
| import org.apache.kafka.connect.data.Struct; | ||
| import org.apache.kafka.connect.transforms.Transformation; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| import java.util.Map; | ||
| import java.util.regex.Matcher; | ||
|
|
||
| public abstract class PatternFilter<R extends ConnectRecord<R>> implements Transformation<R> { | ||
| private static final Logger log = LoggerFactory.getLogger(PatternFilter.class); | ||
|
|
||
| @Override | ||
| public ConfigDef config() { | ||
| return PatternFilterConfig.config(); | ||
| } | ||
|
|
||
| PatternFilterConfig config; | ||
|
|
||
| @Override | ||
| public void configure(Map<String, ?> settings) { | ||
| this.config = new PatternFilterConfig(settings); | ||
| } | ||
|
|
||
| @Override | ||
| public void close() { | ||
|
|
||
| } | ||
|
|
||
| R filter(R record, Struct struct) { | ||
| for (Field field : struct.schema().fields()) { | ||
| if (this.config.fields.contains(field.name())) { | ||
| if (field.schema().type() == Schema.Type.STRING) { | ||
| String input = struct.getString(field.name()); | ||
| if (null != input) { | ||
| Matcher matcher = this.config.pattern.matcher(input); | ||
| if (matcher.matches()) { | ||
| return null; | ||
| } | ||
| } | ||
| } | ||
| } | ||
| } | ||
| return record; | ||
| } | ||
|
|
||
| R filter(R record, Map map) { | ||
| for (Object field : map.keySet()) { | ||
| if (this.config.fields.contains(field)) { | ||
| Object value = map.get(field); | ||
|
|
||
| if (value instanceof String) { | ||
| String input = (String) value; | ||
| Matcher matcher = this.config.pattern.matcher(input); | ||
| if (matcher.matches()) { | ||
| return null; | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| return record; | ||
| } | ||
|
|
||
|
|
||
| R filter(R record, final boolean key) { | ||
| final SchemaAndValue input = key ? | ||
| new SchemaAndValue(record.keySchema(), record.key()) : | ||
| new SchemaAndValue(record.valueSchema(), record.value()); | ||
| final R result; | ||
| if (input.schema() != null) { | ||
| if (Schema.Type.STRUCT == input.schema().type()) { | ||
| result = filter(record, (Struct) input.value()); | ||
| } else if (Schema.Type.MAP == input.schema().type()) { | ||
| result = filter(record, (Map) input.value()); | ||
| } else { | ||
| result = record; | ||
| } | ||
| } else if (input.value() instanceof Map) { | ||
| result = filter(record, (Map) input.value()); | ||
| } else { | ||
| result = record; | ||
| } | ||
|
|
||
| return result; | ||
| } | ||
|
|
||
| @Title("PatternFilter(Key)") | ||
| @Description("This transformation is used to filter records based on a regular expression.") | ||
| @DocumentationTip("This transformation is used to filter records based on fields in the Key of the record.") | ||
| public static class Key<R extends ConnectRecord<R>> extends PatternFilter<R> { | ||
| @Override | ||
| public R apply(R r) { | ||
| return filter(r, true); | ||
| } | ||
| } | ||
|
|
||
| @Title("PatternFilter(Value)") | ||
| @Description("This transformation is used to filter records based on a regular expression.") | ||
| @DocumentationTip("This transformation is used to filter records based on fields in the Value of the record.") | ||
| public static class Value<R extends ConnectRecord<R>> extends PatternFilter<R> { | ||
| @Override | ||
| public R apply(R r) { | ||
| return filter(r, false); | ||
| } | ||
| } | ||
| } |
65 changes: 65 additions & 0 deletions
65
...ain/java/com/github/jcustenborder/kafka/connect/transform/common/PatternFilterConfig.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| /** | ||
| * Copyright © 2017 Jeremy Custenborder ([email protected]) | ||
| * | ||
| * 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 com.github.jcustenborder.kafka.connect.transform.common; | ||
|
|
||
| import com.github.jcustenborder.kafka.connect.utils.config.ConfigKeyBuilder; | ||
| import com.github.jcustenborder.kafka.connect.utils.config.ConfigUtils; | ||
| import com.github.jcustenborder.kafka.connect.utils.config.validators.Validators; | ||
| import org.apache.kafka.common.config.AbstractConfig; | ||
| import org.apache.kafka.common.config.ConfigDef; | ||
|
|
||
| import java.util.Collections; | ||
| import java.util.HashSet; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.Set; | ||
| import java.util.regex.Pattern; | ||
|
|
||
| public class PatternFilterConfig extends AbstractConfig { | ||
| public final Pattern pattern; | ||
| public final Set<String> fields; | ||
|
|
||
| public static final String PATTERN_CONFIG = "pattern"; | ||
| public static final String PATTERN_DOC = "The regex to test the message with. "; | ||
|
|
||
| public static final String FIELD_CONFIG = "fields"; | ||
| public static final String FIELD_DOC = "The fields to transform."; | ||
|
|
||
|
|
||
| public PatternFilterConfig(Map<String, ?> settings) { | ||
| super(config(), settings); | ||
| this.pattern = ConfigUtils.pattern(this, PATTERN_CONFIG); | ||
| List<String> fields = getList(FIELD_CONFIG); | ||
| this.fields = new HashSet<>(fields); | ||
| } | ||
|
|
||
| public static ConfigDef config() { | ||
| return new ConfigDef() | ||
| .define( | ||
| ConfigKeyBuilder.of(PATTERN_CONFIG, ConfigDef.Type.STRING) | ||
| .documentation(PATTERN_DOC) | ||
| .importance(ConfigDef.Importance.HIGH) | ||
| .validator(Validators.pattern()) | ||
| .build() | ||
| ).define( | ||
| ConfigKeyBuilder.of(FIELD_CONFIG, ConfigDef.Type.LIST) | ||
| .documentation(FIELD_DOC) | ||
| .defaultValue(Collections.emptyList()) | ||
| .importance(ConfigDef.Importance.HIGH) | ||
| .build() | ||
| ); | ||
| } | ||
| } |
69 changes: 69 additions & 0 deletions
69
src/test/java/com/github/jcustenborder/kafka/connect/transform/common/PatternFilterTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,69 @@ | ||
| package com.github.jcustenborder.kafka.connect.transform.common; | ||
|
|
||
| import com.google.common.collect.ImmutableMap; | ||
| 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.sink.SinkRecord; | ||
| import org.junit.jupiter.api.BeforeEach; | ||
| import org.junit.jupiter.api.Test; | ||
|
|
||
| import static org.junit.jupiter.api.Assertions.assertNotNull; | ||
| import static org.junit.jupiter.api.Assertions.assertNull; | ||
|
|
||
| public class PatternFilterTest { | ||
| public PatternFilter.Value transform; | ||
|
|
||
| @BeforeEach | ||
| public void before() { | ||
| this.transform = new PatternFilter.Value(); | ||
| this.transform.configure( | ||
| ImmutableMap.of( | ||
| PatternFilterConfig.FIELD_CONFIG, "input", | ||
| PatternFilterConfig.PATTERN_CONFIG, "^filter$" | ||
| ) | ||
| ); | ||
| } | ||
|
|
||
| SinkRecord map(String value) { | ||
| return new SinkRecord( | ||
| "asdf", | ||
| 1, | ||
| null, | ||
| null, | ||
| null, | ||
| ImmutableMap.of("input", value), | ||
| 1234L | ||
| ); | ||
| } | ||
|
|
||
| SinkRecord struct(String value) { | ||
| Schema schema = SchemaBuilder.struct() | ||
| .field("input", Schema.STRING_SCHEMA) | ||
| .build(); | ||
| Struct struct = new Struct(schema) | ||
| .put("input", value); | ||
| return new SinkRecord( | ||
| "asdf", | ||
| 1, | ||
| null, | ||
| null, | ||
| schema, | ||
| struct, | ||
| 1234L | ||
| ); | ||
| } | ||
|
|
||
| @Test | ||
| public void filtered() { | ||
| assertNull(this.transform.apply(struct("filter"))); | ||
| assertNull(this.transform.apply(map("filter"))); | ||
| } | ||
|
|
||
| @Test | ||
| public void notFiltered() { | ||
| assertNotNull(this.transform.apply(struct("ok"))); | ||
| assertNotNull(this.transform.apply(map("ok"))); | ||
| } | ||
|
|
||
| } |