Skip to content

Commit

Permalink
Added a transformation to change a topic name to be all lowercase. Fixes
Browse files Browse the repository at this point in the history
 #92. (#93)

* Added a transformation to change a topic name to be all lowercase. Fixes #92.

* Bumped to use 3.3.1-1

* Added xml-apis.
  • Loading branch information
jcustenborder authored Oct 13, 2022
1 parent 44adce9 commit 12f3084
Show file tree
Hide file tree
Showing 4 changed files with 134 additions and 11 deletions.
34 changes: 23 additions & 11 deletions pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -16,14 +16,15 @@
limitations under the License.
-->
<project xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
<project
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.github.jcustenborder.kafka.connect</groupId>
<artifactId>kafka-connect-parent</artifactId>
<version>2.6.1</version>
<version>3.3.1-1</version>
</parent>
<artifactId>kafka-connect-transform-common</artifactId>
<version>0.1.0-SNAPSHOT</version>
Expand All @@ -49,8 +50,11 @@
</developer>
</developers>
<scm>
<connection>scm:git:https://github.com/jcustenborder/kafka-connect-transform-common.git</connection>
<developerConnection>scm:git:[email protected]:jcustenborder/kafka-connect-transform-common.git</developerConnection>
<connection>scm:git:https://github.com/jcustenborder/kafka-connect-transform-common.git
</connection>
<developerConnection>
scm:git:[email protected]:jcustenborder/kafka-connect-transform-common.git
</developerConnection>
<url>https://github.com/jcustenborder/kafka-connect-transform-common</url>
</scm>
<issueManagement>
Expand All @@ -61,8 +65,10 @@
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>connect-json</artifactId>
<version>${kafka.version}</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>com.github.jcustenborder.kafka.connect</groupId>
<artifactId>connect-utils-jackson</artifactId>
</dependency>
<dependency>
<groupId>org.reflections</groupId>
Expand All @@ -73,7 +79,7 @@
<dependency>
<groupId>com.github.jcustenborder.kafka.connect</groupId>
<artifactId>connect-utils-testing-data</artifactId>
<version>[0.3.33,0.3.1000)</version>
<version>${connect-utils.version}</version>
<scope>test</scope>
</dependency>
<dependency>
Expand All @@ -85,7 +91,12 @@
<groupId>xerces</groupId>
<artifactId>xercesImpl</artifactId>
<version>2.12.1</version>
</dependency>
</dependency>
<dependency>
<groupId>xml-apis</groupId>
<artifactId>xml-apis</artifactId>
<version>1.4.01</version>
</dependency>
</dependencies>
<build>
<plugins>
Expand All @@ -94,7 +105,8 @@
<artifactId>kafka-connect-maven-plugin</artifactId>
<configuration>
<confluentControlCenterIntegration>true</confluentControlCenterIntegration>
<documentationUrl>https://jcustenborder.github.io/kafka-connect-documentation/</documentationUrl>
<documentationUrl>https://jcustenborder.github.io/kafka-connect-documentation/
</documentationUrl>
<componentTypes>
<componentType>transform</componentType>
</componentTypes>
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
/**
* 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.Title;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.utils.SystemTime;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.connect.connector.ConnectRecord;
import org.apache.kafka.connect.transforms.Transformation;

import java.util.Map;

@Title("LowerCaseTopic")
@Description("This transformation is used to change a topic name to be all lower case.")
public class LowerCaseTopic<R extends ConnectRecord<R>> implements Transformation<R> {
Time time = SystemTime.SYSTEM;

@Override
public R apply(R record) {
return record.newRecord(
record.topic().toLowerCase(),
record.kafkaPartition(),
record.keySchema(),
record.key(),
record.valueSchema(),
record.value(),
record.timestamp(),
record.headers()
);
}

@Override
public ConfigDef config() {
return new ConfigDef();
}

@Override
public void close() {

}

@Override
public void configure(Map<String, ?> map) {

}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,35 @@
package com.github.jcustenborder.kafka.connect.transform.common;

import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.connect.sink.SinkRecord;
import org.junit.jupiter.api.Test;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

public class LowerCaseTopicTest {

@Test
public void test() {
final SinkRecord input = new SinkRecord(
"TeSt",
1,
null,
"",
null,
"",
1234123L,
12341312L,
TimestampType.NO_TIMESTAMP_TYPE
);
LowerCaseTopic<SinkRecord> transform = new LowerCaseTopic<>();
final SinkRecord actual = transform.apply(input);
assertEquals("test", actual.topic(), "Topic should match.");
}


}
Original file line number Diff line number Diff line change
@@ -0,0 +1,15 @@
{
"input" : {
"topic" : "TestTopic",
"kafkaPartition" : 1,
"key" : "",
"value" : "",
"timestamp" : 12341312,
"timestampType" : "NO_TIMESTAMP_TYPE",
"offset" : 1234123,
"headers" : [ ]
},
"description" : "This example will change the topic name to be all lower case.",
"name" : "Example",
"config" : { }
}

0 comments on commit 12f3084

Please sign in to comment.