🎉 New Destination: Kafka #3746

mmolimar · 2021-05-30T17:02:44Z

Destination for Apache Kafka.

Related with #1855

Checklist

marcosmarxm · 2021-05-30T18:33:11Z

that's amazing @mmolimar

michel-tricot · 2021-05-31T05:34:24Z

Thank you @mmolimar !!

sherifnada · 2021-06-01T23:29:08Z

@mmolimar incredible, thank you for sharing your connector! Will review very soon -- need to bootstrap my Kafka knowledge. Are you blocked on merging this to Airbyte to be able to use it in your instance?

…fka-destination

subodh1810

@mmolimar This looks great. I am just curious about the setup options that end user will have. Chances are that the end user might not know the ideal value for all the options for their use case. For instance what should be the right Batch size , Buffer memory , Max request size and in that case they might go ahead with the default option. Is the default option ideal for all the use cases? Also can we add more information in the docs about these setup options to help people figure out the right values for these based on their use case

airbyte-integrations/connectors/destination-kafka/src/main/resources/spec.json

…fka-destination

mmolimar · 2021-06-02T13:24:33Z

@mmolimar incredible, thank you for sharing your connector! Will review very soon -- need to bootstrap my Kafka knowledge. Are you blocked on merging this to Airbyte to be able to use it in your instance?

Thanks @sherifnada
Actually I'm not blocked right now but it'd be great to know your thoughts about this to get it merge ;-)

mmolimar · 2021-06-02T13:30:00Z

@mmolimar This looks great. I am just curious about the setup options that end user will have. Chances are that the end user might not know the ideal value for all the options for their use case. For instance what should be the right Batch size , Buffer memory , Max request size and in that case they might go ahead with the default option. Is the default option ideal for all the use cases? Also can we add more information in the docs about these setup options to help people figure out the right values for these based on their use case

Thanks for your comments @subodh1810!
The producer config is very configurable to fit your specific use case (delivery guarantees, throughput, latency, etc.) so there is not an ideal config for all scenarios. The configs set by default are the ones that the Kafka Producer has in its API.
Btw, I added to the README a link with the Kafka Producer configs (not all of them are in the spec but just the most relevant ones).

…fka-destination

marcosmarxm · 2021-06-11T18:14:12Z

Thanks @mmolimar, do you think this is ready to review, or do you need any assistance?

mmolimar · 2021-06-11T18:27:27Z

I think it's fine. Just to resolve this little conflict

…fka-destination

sherifnada

@mmolimar Thanks again for creating the PR. This is going to be huge. Left a review!

...rc/main/java/io/airbyte/integrations/standardtest/destination/DestinationAcceptanceTest.java

airbyte-integrations/connectors/destination-kafka/src/main/resources/spec.json

...tegration/java/io/airbyte/integrations/destination/kafka/KafkaDestinationAcceptanceTest.java

sherifnada · 2021-07-09T09:37:45Z

...tegration/java/io/airbyte/integrations/destination/kafka/KafkaDestinationAcceptanceTest.java

+  protected void setup(TestDestinationEnv testEnv) {
+    kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:6.1.1"));
+    kafka.start();
+    try (var ignored = AdminClient.create(Map.of(


Why do we need this block? Could you add a source code comment to help clarify?

Initially I was creating the topics via the AdminClient but I just removed it.

airbyte-integrations/connectors/destination-kafka/src/main/resources/spec.json

sherifnada · 2021-07-09T10:13:50Z

...ation-kafka/src/main/java/io/airbyte/integrations/destination/kafka/KafkaRecordConsumer.java

+
+  private void sendRecordInTransaction(ProducerRecord<String, JsonNode> record) throws Exception {
+    try {
+      producer.beginTransaction();


what is the value of creating a transaction on a single record? isn't that equivalent to having no transactionality?

Just removed this. Having acks=all and sync=true it's enough

sherifnada · 2021-07-09T10:15:05Z

...ation-kafka/src/main/java/io/airbyte/integrations/destination/kafka/KafkaRecordConsumer.java

+
+  @Override
+  protected void startTracked() {
+    Map<AirbyteStreamNameNamespacePair, String> mapped = catalog.getStreams().stream()


could we refactor this logic into a method which returns topicMap and then write some unit tests for it?

sherifnada · 2021-07-09T10:16:08Z

...ation-kafka/src/main/java/io/airbyte/integrations/destination/kafka/KafkaRecordConsumer.java

+      final JsonNode value = Jsons.jsonNode(ImmutableMap.of(
+          JavaBaseConstants.COLUMN_NAME_AB_ID, key,
+          JavaBaseConstants.COLUMN_NAME_EMITTED_AT, recordMessage.getEmittedAt(),
+          JavaBaseConstants.COLUMN_NAME_DATA, recordMessage.getData()));


if a user is writing to a hardcoded stream, they have no way of knowing which stream this came from. Should we include the stream name in the output record?

Good point! I just added it

...ation-kafka/src/main/java/io/airbyte/integrations/destination/kafka/KafkaRecordConsumer.java

…fka-destination

mmolimar · 2021-07-17T21:19:47Z

Hey @sherifnada !
I've addressed all your comments and made the changes. Let me know your thoughts ;-)

sherifnada

@mmolimar thanks for the great PR! LGTM - just a few suggestions for docs/wording but I think we can release this pretty soon!

airbyte-config/init/src/main/resources/seed/destination_definitions.yaml

airbyte-integrations/connectors/destination-kafka/src/main/resources/spec.json

...ation-kafka/src/main/java/io/airbyte/integrations/destination/kafka/KafkaRecordConsumer.java

docs/integrations/destinations/kafka.md

...ation-kafka/src/main/java/io/airbyte/integrations/destination/kafka/KafkaRecordConsumer.java

airbyte-integrations/connectors/destination-kafka/src/main/resources/spec.json

docs/integrations/destinations/kafka.md

...ation-kafka/src/main/java/io/airbyte/integrations/destination/kafka/KafkaRecordConsumer.java

…tions.yaml Co-authored-by: Sherif A. Nada <snadalive@gmail.com>

…a/io/airbyte/integrations/destination/kafka/KafkaRecordConsumer.java Co-authored-by: Sherif A. Nada <snadalive@gmail.com>

Co-authored-by: Sherif A. Nada <snadalive@gmail.com>

…fka-destination

Co-authored-by: Sherif A. Nada <snadalive@gmail.com>

…te into feature/kafka-destination

…fka-destination

sherifnada · 2021-07-22T02:01:32Z

Thank you @mmolimar ! Amazing contribution 🎉 🎉 🎉 🎉 🎉 🎉

mmolimar added 2 commits May 30, 2021 11:53

Kafka destination

0a17bfc

Adding destination definition

792ff02

auto-assign bot requested review from ChristopheDuong and jrhizor May 30, 2021 17:02

Fix tests

bde98db

marcosmarxm added the community_pr label May 30, 2021

marcosmarxm requested review from sherifnada and subodh1810 and removed request for ChristopheDuong May 30, 2021 18:33

mmolimar added 2 commits June 2, 2021 00:02

Adding name resolver

9d7b178

Merge branch 'master' of github.com:airbytehq/airbyte into feature/ka…

2e8c190

…fka-destination

subodh1810 reviewed Jun 2, 2021

View reviewed changes

airbyte-integrations/connectors/destination-kafka/src/main/resources/spec.json Show resolved Hide resolved

mmolimar added 2 commits June 2, 2021 08:21

Spec configs for DBT and normalization

b23d31d

Merge branch 'master' of github.com:airbytehq/airbyte into feature/ka…

99ba342

…fka-destination

mmolimar added 2 commits June 2, 2021 09:06

Updating docs

f48e170

Merge branch 'master' of github.com:airbytehq/airbyte into feature/ka…

e5cbb61

…fka-destination

Merge branch 'master' of github.com:airbytehq/airbyte into feature/ka…

3361a3f

…fka-destination

github-actions bot added the area/documentation Improvements or additions to documentation label Jun 12, 2021

mmolimar added 2 commits June 13, 2021 12:40

Merge branch 'master' of github.com:airbytehq/airbyte into feature/ka…

06fd84b

…fka-destination

Merge branch 'master' of github.com:airbytehq/airbyte into feature/ka…

756033f

…fka-destination

github-actions bot added the area/connectors Connector related issues label Jun 16, 2021

marcosmarxm assigned subodh1810 Jun 17, 2021

jrhizor removed their request for review June 30, 2021 22:04

sherifnada reviewed Jul 9, 2021

View reviewed changes

marcosmarxm changed the title ~~New destination: Kafka~~ 🎉 New Destination: Kafka Jul 9, 2021

mmolimar added 8 commits July 13, 2021 18:37

Merge branch 'master' of github.com:airbytehq/airbyte into feature/ka…

b3565b9

…fka-destination

Adding suggestions

ed8a743

Merge branch 'master' of github.com:airbytehq/airbyte into feature/ka…

9487b0c

…fka-destination

Fix tests

f13c93c

Updating spec

187517d

Flush messages

c132b63

Improvements in spec

dc53912

Add state to the output collector

f99546f

Adding Kafka icon

33ffe98

sherifnada approved these changes Jul 20, 2021

View reviewed changes

mmolimar and others added 9 commits July 20, 2021 16:10

Update airbyte-config/init/src/main/resources/seed/destination_defini…

fd0bd35

…tions.yaml Co-authored-by: Sherif A. Nada <snadalive@gmail.com>

Update airbyte-integrations/connectors/destination-kafka/src/main/jav…

068fc37

…a/io/airbyte/integrations/destination/kafka/KafkaRecordConsumer.java Co-authored-by: Sherif A. Nada <snadalive@gmail.com>

Update docs/integrations/destinations/kafka.md

85b306b

Co-authored-by: Sherif A. Nada <snadalive@gmail.com>

Merge branch 'master' of github.com:airbytehq/airbyte into feature/ka…

c0e5acb

…fka-destination

Update docs/integrations/destinations/kafka.md

ee262d5

Co-authored-by: Sherif A. Nada <snadalive@gmail.com>

Merge branch 'feature/kafka-destination' of github.com:mmolimar/airby…

afc4ea8

…te into feature/kafka-destination

Updating docs

430f37a

format

56ef956

Merge branch 'master' of github.com:airbytehq/airbyte into feature/ka…

e15b971

…fka-destination

sherifnada linked an issue Jul 22, 2021 that may be closed by this pull request

New Destination: Kafka #1855

Closed

sherifnada mentioned this pull request Jul 22, 2021

Publish Kafka Destination #4900

Merged

sherifnada merged commit fc3c692 into airbytehq:master Jul 22, 2021

karinakuz added connectors/destinations-streaming connectors/destination/kafka labels Jan 14, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

🎉 New Destination: Kafka #3746

🎉 New Destination: Kafka #3746

mmolimar commented May 30, 2021 •

edited by sherifnada

Loading

marcosmarxm commented May 30, 2021

michel-tricot commented May 31, 2021

sherifnada commented Jun 1, 2021

subodh1810 left a comment

mmolimar commented Jun 2, 2021

mmolimar commented Jun 2, 2021

marcosmarxm commented Jun 11, 2021

mmolimar commented Jun 11, 2021

sherifnada left a comment

sherifnada Jul 9, 2021

mmolimar Jul 17, 2021

sherifnada Jul 9, 2021

mmolimar Jul 17, 2021

sherifnada Jul 9, 2021

mmolimar Jul 17, 2021

sherifnada Jul 9, 2021

mmolimar Jul 17, 2021

mmolimar commented Jul 17, 2021 •

edited

Loading

sherifnada left a comment

sherifnada commented Jul 22, 2021

🎉 New Destination: Kafka #3746

🎉 New Destination: Kafka #3746

Conversation

mmolimar commented May 30, 2021 • edited by sherifnada Loading

Checklist

marcosmarxm commented May 30, 2021

michel-tricot commented May 31, 2021

sherifnada commented Jun 1, 2021

subodh1810 left a comment

Choose a reason for hiding this comment

mmolimar commented Jun 2, 2021

mmolimar commented Jun 2, 2021

marcosmarxm commented Jun 11, 2021

mmolimar commented Jun 11, 2021

sherifnada left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mmolimar commented Jul 17, 2021 • edited Loading

sherifnada left a comment

Choose a reason for hiding this comment

sherifnada commented Jul 22, 2021

mmolimar commented May 30, 2021 •

edited by sherifnada

Loading

mmolimar commented Jul 17, 2021 •

edited

Loading