-
Notifications
You must be signed in to change notification settings - Fork 4.2k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Source:Kafka - Avro format message support (#15827)
* new format is added * Avro support Avro support * new format is added * schema namespace updated * multi topic schema name is added * max_records_process param is added * review 1 * Schema registry details are added * note update * auto-bump connector version [ci skip] Co-authored-by: Sajarin <sajarindider@gmail.com> Co-authored-by: Octavia Squidington III <octavia-squidington-iii@users.noreply.github.com>
- Loading branch information
3 people
authored
Sep 13, 2022
1 parent
3dfe362
commit ce64e96
Showing
16 changed files
with
690 additions
and
164 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
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
31 changes: 31 additions & 0 deletions
31
...s/source-kafka/src/main/java/io/airbyte/integrations/source/kafka/KafkaFormatFactory.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,31 @@ | ||
| /* | ||
| * Copyright (c) 2022 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.integrations.source.kafka; | ||
|
|
||
| import com.fasterxml.jackson.databind.JsonNode; | ||
| import io.airbyte.integrations.source.kafka.format.AvroFormat; | ||
| import io.airbyte.integrations.source.kafka.format.JsonFormat; | ||
| import io.airbyte.integrations.source.kafka.format.KafkaFormat; | ||
|
|
||
| public class KafkaFormatFactory { | ||
|
|
||
| public static KafkaFormat getFormat(final JsonNode config) { | ||
|
|
||
| MessageFormat messageFormat = | ||
| config.has("MessageFormat") ? MessageFormat.valueOf(config.get("MessageFormat").get("deserialization_type").asText().toUpperCase()) | ||
| : MessageFormat.JSON; | ||
|
|
||
| switch (messageFormat) { | ||
| case JSON -> { | ||
| return new JsonFormat(config); | ||
| } | ||
| case AVRO -> { | ||
| return new AvroFormat(config); | ||
| } | ||
| } | ||
| return new JsonFormat(config); | ||
| } | ||
|
|
||
| } |
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
35 changes: 35 additions & 0 deletions
35
...ectors/source-kafka/src/main/java/io/airbyte/integrations/source/kafka/KafkaStrategy.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,35 @@ | ||
| /* | ||
| * Copyright (c) 2022 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.integrations.source.kafka; | ||
|
|
||
| import io.confluent.kafka.serializers.subject.RecordNameStrategy; | ||
| import io.confluent.kafka.serializers.subject.TopicNameStrategy; | ||
| import io.confluent.kafka.serializers.subject.TopicRecordNameStrategy; | ||
|
|
||
| /** | ||
| * https://docs.confluent.io/platform/current/schema-registry/serdes-develop/index.html | ||
| */ | ||
| public enum KafkaStrategy { | ||
|
|
||
| TopicNameStrategy(TopicNameStrategy.class.getName()), | ||
| RecordNameStrategy(RecordNameStrategy.class.getName()), | ||
| TopicRecordNameStrategy(TopicRecordNameStrategy.class.getName()); | ||
|
|
||
| String className; | ||
|
|
||
| KafkaStrategy(String name) { | ||
| this.className = name; | ||
| } | ||
|
|
||
| public static String getStrategyName(String name) { | ||
| for (KafkaStrategy value : KafkaStrategy.values()) { | ||
| if (value.name().equalsIgnoreCase(name)) { | ||
| return value.className; | ||
| } | ||
| } | ||
| throw new IllegalArgumentException("Unexpected data to strategy setting: " + name); | ||
| } | ||
|
|
||
| } |
14 changes: 14 additions & 0 deletions
14
...ectors/source-kafka/src/main/java/io/airbyte/integrations/source/kafka/MessageFormat.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,14 @@ | ||
| /* | ||
| * Copyright (c) 2022 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.integrations.source.kafka; | ||
|
|
||
| /** | ||
| * message format in kafka queue | ||
| * https://docs.confluent.io/platform/current/schema-registry/serdes-develop/index.html | ||
| */ | ||
| public enum MessageFormat { | ||
| JSON, | ||
| AVRO | ||
| } |
Oops, something went wrong.