-
Notifications
You must be signed in to change notification settings - Fork 5
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #367 from jensur77/main
Support custom `TimeMapper` class
- Loading branch information
Showing
8 changed files
with
257 additions
and
80 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
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
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
19 changes: 19 additions & 0 deletions
19
src/main/java/software/amazon/event/kafkaconnector/mapping/DefaultTimeMapper.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,19 @@ | ||
| /* | ||
| * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
| package software.amazon.event.kafkaconnector.mapping; | ||
|
|
||
| import java.time.Instant; | ||
| import org.apache.kafka.connect.sink.SinkRecord; | ||
|
|
||
| public class DefaultTimeMapper implements TimeMapper { | ||
|
|
||
| @Override | ||
| public Instant getTime(SinkRecord sinkRecord) { | ||
| // As described in AWS documentation | ||
| // https://docs.aws.amazon.com/eventbridge/latest/APIReference/API_PutEventsRequestEntry.html | ||
| // If no timestamp is provided, the timestamp of the PutEvents call is used. | ||
| return null; | ||
| } | ||
| } |
89 changes: 89 additions & 0 deletions
89
src/main/java/software/amazon/event/kafkaconnector/mapping/SinkRecordJsonMapper.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,89 @@ | ||
| /* | ||
| * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
| package software.amazon.event.kafkaconnector.mapping; | ||
|
|
||
| import static java.util.Collections.singletonMap; | ||
|
|
||
| import com.fasterxml.jackson.databind.JsonNode; | ||
| import com.fasterxml.jackson.databind.ObjectMapper; | ||
| import com.fasterxml.jackson.databind.node.ArrayNode; | ||
| import java.io.IOException; | ||
| import org.apache.kafka.connect.header.Header; | ||
| import org.apache.kafka.connect.json.JsonConverter; | ||
| import org.apache.kafka.connect.sink.SinkRecord; | ||
|
|
||
| public class SinkRecordJsonMapper { | ||
| private final JsonConverter jsonConverter = new JsonConverter(); | ||
| private final ObjectMapper objectMapper = new ObjectMapper(); | ||
|
|
||
| public SinkRecordJsonMapper() { | ||
| jsonConverter.configure(singletonMap("schemas.enable", "false"), false); | ||
| } | ||
|
|
||
| public String createJsonPayload(SinkRecord sinkRecord) throws IOException { | ||
| var root = objectMapper.createObjectNode(); | ||
| root.put("topic", sinkRecord.topic()); | ||
| root.put("partition", sinkRecord.kafkaPartition()); | ||
| root.put("offset", sinkRecord.kafkaOffset()); | ||
| root.put("timestamp", sinkRecord.timestamp()); | ||
| root.put("timestampType", sinkRecord.timestampType().toString()); | ||
| root.set("headers", createHeaderArray(sinkRecord)); | ||
|
|
||
| if (sinkRecord.key() == null) { | ||
| root.set("key", null); | ||
| } else { | ||
| root.set( | ||
| "key", | ||
| createJSONFromByteArray( | ||
| jsonConverter.fromConnectData( | ||
| sinkRecord.topic(), sinkRecord.keySchema(), sinkRecord.key()))); | ||
| } | ||
|
|
||
| // tombstone handling | ||
| if (sinkRecord.value() == null) { | ||
| root.set("value", null); | ||
| } else { | ||
| root.set( | ||
| "value", | ||
| createJSONFromByteArray( | ||
| jsonConverter.fromConnectData( | ||
| sinkRecord.topic(), sinkRecord.valueSchema(), sinkRecord.value()))); | ||
| } | ||
| return root.toString(); | ||
| } | ||
|
|
||
| /** | ||
| * This method serializes Kafka message headers to JSON. | ||
| * | ||
| * @param sinkRecord Kafka record to be sent to EventBridge | ||
| * @return headers to be added to EventBridge message | ||
| * @throws IOException | ||
| */ | ||
| private ArrayNode createHeaderArray(SinkRecord sinkRecord) throws IOException { | ||
| var headersArray = objectMapper.createArrayNode(); | ||
|
|
||
| for (Header header : sinkRecord.headers()) { | ||
| var headerItem = objectMapper.createObjectNode(); | ||
| headerItem.set( | ||
| header.key(), | ||
| createJSONFromByteArray( | ||
| jsonConverter.fromConnectHeader( | ||
| sinkRecord.topic(), header.key(), header.schema(), header.value()))); | ||
| headersArray.add(headerItem); | ||
| } | ||
| return headersArray; | ||
| } | ||
|
|
||
| /** | ||
| * This method converts the byteArray which is returned by the {@link JsonConverter} to JSON. | ||
| * | ||
| * @param jsonBytes - byteArray to convert to JSON | ||
| * @return the JSON representation of jsonBytes | ||
| * @throws IOException | ||
| */ | ||
| private JsonNode createJSONFromByteArray(byte[] jsonBytes) throws IOException { | ||
| return objectMapper.readTree(jsonBytes); | ||
| } | ||
| } |
12 changes: 12 additions & 0 deletions
12
src/main/java/software/amazon/event/kafkaconnector/mapping/TimeMapper.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,12 @@ | ||
| /* | ||
| * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. | ||
| * SPDX-License-Identifier: Apache-2.0 | ||
| */ | ||
| package software.amazon.event.kafkaconnector.mapping; | ||
|
|
||
| import java.time.Instant; | ||
| import org.apache.kafka.connect.sink.SinkRecord; | ||
|
|
||
| public interface TimeMapper { | ||
| Instant getTime(SinkRecord sinkRecord); | ||
| } |
Oops, something went wrong.