kafka-connect-transform-common - GitLite

KO | EN

gitlite — search

#javascript #typescript #llm #golang #linux #docker #java #windows #awesome #vue #security #mcp #openai #open-source

kafka-connect-transform-common

(★ 174)

Common Transforms for Kafka Connect.

파일 탐색기

최종 버전 다운로드 (.zip)

- - maven.yml
- PatternRename.conf
- transformations.rst
- validate-cp-8.3.0-plugins.sh
.gitignore
AGENTS.md
docker-compose.yml
LICENSE
pom.xml
README.md

# 설치 가이드

1. 코드 내려받기

git clone https://github.com/jcustenborder/kafka-connect-transform-common

깃허브에서 프로젝트 코드 전체를 내 컴퓨터로 내려받습니다.

cd kafka-connect-transform-common

방금 내려받은 프로젝트 폴더 안으로 이동합니다.

2. Docker

쉬움 추천

사전 준비물

Git GitHub에서 프로젝트 코드를 내려받으려면 필요합니다.
Docker Desktop 컨테이너를 빌드하고 실행하려면 필요합니다. 설치 후 실행해서 백그라운드에 켜두세요.

docker compose up -d --build

정의된 모든 컨테이너(서버, DB 등)를 한 번에 빌드하고 백그라운드에서 실행합니다.

✅ 터미널에 docker compose ps 를 입력해 컨테이너들이 Up 상태인지 확인하세요. README에 포트 번호가 적혀있다면 브라우저에서 http://localhost:포트번호 로 접속해보세요.

3. Maven (Java)

보통

사전 준비물

Git GitHub에서 프로젝트 코드를 내려받으려면 필요합니다.
JDK (Java) Java 프로젝트를 빌드/실행하려면 필요합니다.
Maven mvn 명령어를 쓰기 위한 빌드 도구입니다.

mvn clean package

Maven으로 의존성 설치와 빌드를 진행합니다.

✅ BUILD SUCCESS 메시지가 뜨면 성공입니다. target/ 폴더에 결과물이 생성됩니다.

이 레포의 README에 적힌 실제 명령어를 그대로 가져왔습니다.

// repository documentation

Was this content helpful?

★ 0 (0 ratings)

# Introduction
[Documentation](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-transform-common) | [Confluent Hub](https://www.confluent.io/hub/jcustenborder/kafka-connect-transform-common)

This project contains common transformations for every day use cases with Kafka Connect.

# Installation

## Confluent Hub

The following command can be used to install the plugin directly from the Confluent Hub using the
[Confluent Hub Client](https://docs.confluent.io/current/connect/managing/confluent-hub/client.html).

```bash
confluent-hub install jcustenborder/kafka-connect-transform-common:latest
```

## Manually

The zip file that is deployed to the [Confluent Hub](https://www.confluent.io/hub/jcustenborder/kafka-connect-transform-common) is available under
`target/components/packages/`. You can manually extract this zip file which includes all dependencies. All the dependencies
that are required to deploy the plugin are under `target/kafka-connect-target` as well. Make sure that you include all the dependencies that are required
to run the plugin.

1. Create a directory under the `plugin.path` on your Connect worker.
2. Copy all of the dependencies under the newly created subdirectory.
3. Restart the Connect worker.

# Transformations
## [BytesToString](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-transform-common/transformations/BytesToString.html)

*Key*
```
com.github.jcustenborder.kafka.connect.transform.common.BytesToString$Key
```
*Value*
```
com.github.jcustenborder.kafka.connect.transform.common.BytesToString$Value
```

### Configuration

#### General

##### `charset`

The charset to use when creating the output string.

*Importance:* HIGH

*Type:* STRING

*Default Value:* UTF-8

##### `fields`

The fields to transform.

*Importance:* HIGH

*Type:* LIST

## [ChangeCase](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-transform-common/transformations/ChangeCase.html)

*Key*
```
com.github.jcustenborder.kafka.connect.transform.common.ChangeCase$Key
```
*Value*
```
com.github.jcustenborder.kafka.connect.transform.common.ChangeCase$Value
```

### Configuration

#### General

##### `from`

The format to move from

*Importance:* HIGH

*Type:* STRING

*Validator:* Matches: ``LOWER_HYPHEN``, ``LOWER_UNDERSCORE``, ``LOWER_CAMEL``, ``UPPER_CAMEL``, ``UPPER_UNDERSCORE``

##### `to`

*Importance:* HIGH

*Type:* STRING

*Validator:* Matches: ``LOWER_HYPHEN``, ``LOWER_UNDERSCORE``, ``LOWER_CAMEL``, ``UPPER_CAMEL``, ``UPPER_UNDERSCORE``

## [ChangeTopicCase](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-transform-common/transformations/ChangeTopicCase.html)

```
com.github.jcustenborder.kafka.connect.transform.common.ChangeTopicCase
```

This transformation is used to change the case of a topic.

[✍️ Example](https://rmoff.net/2020/12/23/twelve-days-of-smt-day-12-community-transformations/#_change_the_topic_case) / [🎥 Video](https://www.youtube.com/watch?v=Z7k_6vGRrkc&t=274s)

### Tip

This transformation will convert a topic name like 'TOPIC_NAME' to `topicName`, or `topic_name`.
### Configuration

#### General

##### `from`

The format of the incoming topic name. `LOWER_CAMEL` = Java variable naming convention, e.g., "lowerCamel". `LOWER_HYPHEN` = Hyphenated variable naming convention, e.g., "lower-hyphen". `LOWER_UNDERSCORE` = C++ variable naming convention, e.g., "lower_underscore". `UPPER_CAMEL` = Java and C++ class naming convention, e.g., "UpperCamel". `UPPER_UNDERSCORE` = Java and C++ constant naming convention, e.g., "UPPER_UNDERSCORE".

*Importance:* HIGH

*Type:* STRING

*Validator:* Matches: ``LOWER_HYPHEN``, ``LOWER_UNDERSCORE``, ``LOWER_CAMEL``, ``UPPER_CAMEL``, ``UPPER_UNDERSCORE``

##### `to`

The format of the outgoing topic name. `LOWER_CAMEL` = Java variable naming convention, e.g., "lowerCamel". `LOWER_HYPHEN` = Hyphenated variable naming convention, e.g., "lower-hyphen". `LOWER_UNDERSCORE` = C++ variable naming convention, e.g., "lower_underscore". `UPPER_CAMEL` = Java and C++ class naming convention, e.g., "UpperCamel". `UPPER_UNDERSCORE` = Java and C++ constant naming convention, e.g., "UPPER_UNDERSCORE".

*Importance:* HIGH

*Type:* STRING

*Validator:* Matches: ``LOWER_HYPHEN``, ``LOWER_UNDERSCORE``, ``LOWER_CAMEL``, ``UPPER_CAMEL``, ``UPPER_UNDERSCORE``

## [ExtractNestedField](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-transform-common/transformations/ExtractNestedField.html)

*Key*
```
com.github.jcustenborder.kafka.connect.transform.common.ExtractNestedField$Key
```
*Value*
```
com.github.jcustenborder.kafka.connect.transform.common.ExtractNestedField$Value
```

### Configuration

#### General

##### `input.inner.field.name`

The field on the child struct containing the field to be extracted. For example if you wanted the extract `address.state` you would use `state`.

*Importance:* HIGH

*Type:* STRING

##### `input.outer.field.name`

The field on the parent struct containing the child struct. For example if you wanted the extract `address.state` you would use `address`.

*Importance:* HIGH

*Type:* STRING

##### `output.field.name`

The field to place the extracted value into.

*Importance:* HIGH

*Type:* STRING

## [ExtractTimestamp](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-transform-common/transformations/ExtractTimestamp.html)

*Key*
```
com.github.jcustenborder.kafka.connect.transform.common.ExtractTimestamp$Key
```
*Value*
```
com.github.jcustenborder.kafka.connect.transform.common.ExtractTimestamp$Value
```

This transformation is used to use a field from the input data to override the timestamp for the record.

[✍️ Example](https://rmoff.net/2020/12/23/twelve-days-of-smt-day-12-community-transformations/#_add_the_timestamp_of_a_field_to_the_topic_name) / [🎥 Video](https://www.youtube.com/watch?v=Z7k_6vGRrkc&t=430s)

### Configuration

#### General

##### `field.name`

The field to pull the timestamp from. This must be an int64 or a timestamp.

*Importance:* HIGH

*Type:* STRING

## [HeaderToField](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-transform-common/transformations/HeaderToField.html)

*Key*
```
com.github.jcustenborder.kafka.connect.transform.common.HeaderToField$Key
```
*Value*
```
com.github.jcustenborder.kafka.connect.transform.common.HeaderToField$Value
```

### Configuration

#### General

##### `header.mappings`

The mapping of the header to the field in the message.

*Importance:* HIGH

*Type:* LIST

## [NormalizeSchema](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-transform-common/transformations/NormalizeSchema.html)

*Key*
```
com.github.jcustenborder.kafka.connect.transform.common.NormalizeSchema$Key
```
*Value*
```
com.github.jcustenborder.kafka.connect.transform.common.NormalizeSchema$Value
```

This transformation is used to convert older schema versions to the latest schema version. This works by keying all of the schemas that are coming into the transformation by their schema name and comparing the version() of the schema. The latest version of a schema will be used. Schemas are discovered as the flow through the transformation. The latest version of a schema is what is used.
### Configuration

## [PatternFilter](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-transform-common/transformations/PatternFilter.html)

*Key*
```
com.github.jcustenborder.kafka.connect.transform.common.PatternFilter$Key
```
*Value*
```
com.github.jcustenborder.kafka.connect.transform.common.PatternFilter$Value
```

### Configuration

#### General

##### `pattern`

The regex to test the message with.

*Importance:* HIGH

*Type:* STRING

*Validator:* com.github.jcustenborder.kafka.connect.utils.config.validators.PatternValidator@4170ee0f

##### `fields`

The fields to transform.

*Importance:* HIGH

*Type:* LIST

## [PatternRename](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-transform-common/transformations/PatternRename.html)

*Key*
```
com.github.jcustenborder.kafka.connect.transform.common.PatternRename$Key
```
*Value*
```
com.github.jcustenborder.kafka.connect.transform.common.PatternRename$Value
```

### Configuration

#### General

##### `field.pattern`

*Importance:* HIGH

*Type:* STRING

##### `field.replacement`

*Importance:* HIGH

*Type:* STRING

##### `field.pattern.flags`

*Importance:* LOW

*Type:* LIST

*Default Value:* [CASE_INSENSITIVE]

*Validator:* [UNICODE_CHARACTER_CLASS, CANON_EQ, UNICODE_CASE, DOTALL, LITERAL, MULTILINE, COMMENTS, CASE_INSENSITIVE, UNIX_LINES]

## [SchemaNameToTopic](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-transform-common/transformations/SchemaNameToTopic.html)

*Key*
```
com.github.jcustenborder.kafka.connect.transform.common.SchemaNameToTopic$Key
```
*Value*
```
com.github.jcustenborder.kafka.connect.transform.common.SchemaNameToTopic$Value
```

This transformation is used to take the name from the schema for the key or value and replace the topic with this value.
### Configuration

## [SetMaximumPrecision](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-transform-common/transformations/SetMaximumPrecision.html)

*Key*
```
com.github.jcustenborder.kafka.connect.transform.common.SetMaximumPrecision$Key
```
*Value*
```
com.github.jcustenborder.kafka.connect.transform.common.SetMaximumPrecision$Value
```

This transformation is used to ensure that all decimal fields in a struct are below the maximum precision specified.
### Note

The Confluent AvroConverter uses a default precision of 64 which can be too large for some database systems.
### Configuration

#### General

##### `precision.max`

The maximum precision allowed.

*Importance:* HIGH

*Type:* INT

*Validator:* [1,...,64]

## [SetNull](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-transform-common/transformations/SetNull.html)

*Key*
```
com.github.jcustenborder.kafka.connect.transform.common.SetNull$Key
```
*Value*
```
com.github.jcustenborder.kafka.connect.transform.common.SetNull$Value
```

### Configuration

## [TimestampNow](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-transform-common/transformations/TimestampNow.html)

```
com.github.jcustenborder.kafka.connect.transform.common.TimestampNow
```

This transformation is used to override the timestamp of the incoming record to the time the record is being processed.
### Configuration

## [TimestampNowField](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-transform-common/transformations/TimestampNowField.html)

*Key*
```
com.github.jcustenborder.kafka.connect.transform.common.TimestampNowField$Key
```
*Value*
```
com.github.jcustenborder.kafka.connect.transform.common.TimestampNowField$Value
```

This transformation is used to set a field with the current timestamp of the system running the transformation.

[✍️ Example](https://rmoff.net/2020/12/23/twelve-days-of-smt-day-12-community-transformations/#_add_the_current_timestamp_to_the_message_payload) / [🎥 Video](https://www.youtube.com/watch?v=Z7k_6vGRrkc&t=679s)

### Configuration

#### General

##### `fields`

The field(s) that will be inserted with the timestamp of the system.

*Importance:* HIGH

*Type:* LIST

## [ToJSON](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-transform-common/transformations/ToJSON.html)

*Key*
```
com.github.jcustenborder.kafka.connect.transform.common.ToJSON$Key
```
*Value*
```
com.github.jcustenborder.kafka.connect.transform.common.ToJSON$Value
```

### Configuration

#### General

##### `output.schema.type`

The connect schema type to output the converted JSON as.

*Importance:* MEDIUM

*Type:* STRING

*Default Value:* STRING

*Validator:* [STRING, BYTES]

##### `schemas.enable`

Flag to determine if the JSON data should include the schema.

*Importance:* MEDIUM

*Type:* BOOLEAN

## [ToLong](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-transform-common/transformations/ToLong.html)

*Key*
```
com.github.jcustenborder.kafka.connect.transform.common.ToLong$Key
```
*Value*
```
com.github.jcustenborder.kafka.connect.transform.common.ToLong$Value
```

### Configuration

#### General

##### `fields`

The fields to transform.

*Importance:* HIGH

*Type:* LIST

## [TopicNameToField](https://jcustenborder.github.io/kafka-connect-documentation/projects/kafka-connect-transform-common/transformations/TopicNameToField.html)

*Key*
```
com.github.jcustenborder.kafka.connect.transform.common.TopicNameToField$Key
```
*Value*
```
com.github.jcustenborder.kafka.connect.transform.common.TopicNameToField$Value
```

### Configuration

#### General

##### `field`

The field to insert the topic name.

*Importance:* HIGH

*Type:* STRING

# Development

## Building the source

```bash
mvn clean package
```

## Contributions

Contributions are always welcomed! Before you start any development please create an issue and
start a discussion. Create a pull request against your newly created issue and we're happy to see
if we can merge your pull request. First and foremost any time you're adding code to the code base
you need to include test coverage. Make sure that you run `mvn clean package` before submitting your
pull to ensure that all of the tests, checkstyle rules, and the package can be successfully built.