feat: support initialisation of producer without topic for kafka, azure event hub and confluent cloud #2569

utsabc · 2022-10-14T05:11:10Z

Description

The scope of this PR is to support KAFKA, Azure event hub and confluent cloud producer with multiple topic from a single client. We want to generate the kafka client without the context of producer where in we will be passing topic information at individual message level.

Transformer takes the responsibility of allocating correct topic information to messages

To support backward compatibility the server client falls back to default topic present in destination config if the message level topic is not present.

Also from transformer to support fair event ordering for topics the logic that is added are as follows

events --> server --> transformer (transform each message) ---> server ---> client --> delivery
[rudderId = anonymousId <<>>userID ] [rudderId = anonymousId<<>>userId<<>>topic]

Notion Ticket

Security

The code changed/added as part of this pull request won't create any security issues with how the software is being used.

codecov · 2022-10-14T05:25:21Z

Codecov Report

Base: 43.73% // Head: 43.71% // Decreases project coverage by -0.01% ⚠️

Coverage data is based on head (9f7e642) compared to base (8ac05c2).
Patch coverage: 71.42% of modified lines in pull request are covered.

❗ Current head 9f7e642 differs from pull request most recent head eeb8667. Consider uploading reports for the commit eeb8667 to get more accurate results

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #2569      +/-   ##
==========================================
- Coverage   43.73%   43.71%   -0.02%     
==========================================
  Files         191      191              
  Lines       40483    40499      +16     
==========================================
- Hits        17707    17706       -1     
- Misses      21672    21685      +13     
- Partials     1104     1108       +4

Impacted Files	Coverage Δ
services/streammanager/kafka/client/producer.go	`78.50% <33.33%> (-1.89%)`	⬇️
services/streammanager/kafka/kafkamanager.go	`64.16% <100.00%> (+0.94%)`	⬆️
enterprise/reporting/setup.go	`57.14% <0.00%> (-14.29%)`	⬇️
services/stats/stats.go	`88.48% <0.00%> (-1.82%)`	⬇️
enterprise/reporting/reporting.go	`8.33% <0.00%> (-1.44%)`	⬇️
services/rsources/handler.go	`69.72% <0.00%> (-0.84%)`	⬇️

Help us with your feedback. Take ten seconds to tell us how you rate us. Have a feature suggestion? Share it here.

☔ View full report at Codecov.
📢 Do you have feedback about the report comment? Let us know in this issue.

…nor refactors

services/streammanager/kafka/client/producer.go

services/streammanager/kafka/kafkamanager.go

services/streammanager/kafka/client/producer.go

krishna2020 · 2022-10-19T11:22:19Z

Notion link: https://www.notion.so/rudderstacks/Kafka-Destination-with-Multiple-Topics-85cbce26706d49eca6853b972e5497b0

atzoum

We need some test cases for all the new scenarios introduced for client.Producer:

Initialize a producer with a topic and send a message containing a topic, the message's topic shall be ignored
Initialize a producer without a topic and try to send a message containing no topic, an error should be returned

Furthermore, we need test scenarios for kafkamanager as well covering

Reading the topic from the message
Falling back to the default topic if no topic is provided by the message
Handling invalid topic fields (e.g. number, object, nil)

atzoum · 2022-10-20T07:04:23Z

services/streammanager/kafka/client/producer.go

@@ -132,6 +124,19 @@ func (p *Producer) Publish(ctx context.Context, msgs ...Message) error {

 var tempError interface{ Temporary() bool }

+func processHeaderForEachMessage(msg Message) (headers []kafka.Header) {


minor

Suggested change

func processHeaderForEachMessage(msg Message) (headers []kafka.Header) {

func headers(msg Message) (headers []kafka.Header) {

This is refactored and addressed

atzoum · 2022-10-20T07:04:35Z

services/streammanager/kafka/client/producer.go

-				}
-			}
-		}
+		headers := processHeaderForEachMessage(msgs[i])


Suggested change

headers := processHeaderForEachMessage(msgs[i])

headers := headers(msgs[i])

This is refactored and addressed

atzoum · 2022-10-20T07:09:40Z

services/streammanager/kafka/client/producer.go

 		messages[i] = kafka.Message{
+			Topic:   msgs[i].Topic,


client.Producer should know if it has been initialized with a topic or not and only include the message's topic in the latter case.
Additionally, this function should return an error if the producer is not initialized with a topic and no topic is included in at least one of the messages.

We plan to go with the approach where we always initialize the producer without client and we intend to always send each message with topic information. That proper assignment of topic to each message would be handled at transformer

Given that we decided to go down this road, can you please make it mandatory and raise an error here if the topic is empty?

for i := range msgs { if msgs[i].Topic == "" { return fmt.Errorf("no topic provided for message %d", i) }

I think we should also have a test for this (don't specify topic and check that an error is returned).

atzoum · 2022-10-20T07:14:35Z

services/streammanager/kafka/kafkamanager.go

 	for i, data := range batch {
+		if conf.MultiTopicSupport {
+			topic = data["topic"].(string)


shouldn't we fallback to the default topic if the message doesn't contain one?

This is handled with transformer being the responsible service for assigning to topic to each message

opensource users don't always update their transformers, do they? Plus I don't feel safe with this decision (have commented about it elswhere)

atzoum · 2022-10-20T07:17:15Z

services/streammanager/kafka/kafkamanager.go

@@ -611,7 +617,11 @@ func sendMessage(ctx context.Context, jsonData json.RawMessage, p producerManage
 			return makeErrorResponse(fmt.Errorf("unable to serialize event with messageId: %s, with error %s", messageId, err))
 		}
 	}
+	if config.MultiTopicSupport {
+		topic = parsedJSON.Get("topic").Value().(string)


type cast will panic here if the field is absent or not a string. Additionally, shouldn't we fallback to the default topic if the message doesn't contain one?

Handled similar vulnerabilities

fracasula · 2022-10-20T11:16:34Z

services/streammanager/kafka/kafkamanager.go

@@ -312,7 +313,7 @@ func NewProducer(destination *backendconfig.DestinationT, o common.Opts) (*Produ
 		return nil, fmt.Errorf("could not ping: %w", err)
 	}

-	p, err := c.NewProducer(destConfig.Topic, client.ProducerConfig{
+	p, err := c.NewProducer("", client.ProducerConfig{


If this is how you want to handle topics then why keeping the argument in the constructor? It looks like you'll be passing it each time anyway even for old configurations.

Yes we are planning the same approach updated for the same

fracasula · 2022-10-20T11:23:59Z

We need some test cases for all the new scenarios introduced for client.Producer:

Initialize a producer with a topic and send a message containing a topic, the message's topic shall be ignored

Initialize a producer without a topic and try to send a message containing no topic, an error should be returned

To simplify a bit I'd remove the topic argument from NewProducer and always demand it on Publish. I'm suggesting this because of the approach proposed in this PR it looks like we would always pass it on Publish anyway so we can clean up the constructor and have only a single scenario to test there.

services/streammanager/kafka/client/producer.go

…removing client initialisation with topic

…values from message

koladilip

Looks like tests are failing, check once, also make sure to add unit test cases for the code you have modified

services/streammanager/kafka/kafkamanager.go

…bility support for default topic

services/streammanager/kafka/client/producer.go

services/streammanager/kafka/kafkamanager.go

atzoum · 2022-11-07T08:41:29Z

@utsabc if this is actually the final product, can you please take care of the PR title and description accordingly?

utsabc · 2022-11-07T08:56:57Z

@utsabc if this is actually the final product, can you please take care of the PR title and description accordingly?

Addressed

services/streammanager/kafka/kafkamanager_test.go

fracasula · 2022-11-07T10:02:58Z

services/streammanager/kafka/kafkamanager.go

@@ -450,14 +450,19 @@ func serializeAvroMessage(value []byte, codec goavro.Codec) ([]byte, error) {
 	return binary, nil
 }

-func prepareBatchOfMessages(topic string, batch []map[string]interface{}, timestamp time.Time, p producerManager) (
+func prepareBatchOfMessages(batch []map[string]interface{}, timestamp time.Time, p producerManager, defaultTopic string) (


Instead of passing the defaultTopic around, an alternative approach is to use the existing pattern of including such information in the producerManager:

type producerManager interface { io.Closer publisher getDefaultTopic() string getTimeout() time.Duration getCodecs() map[string]*goavro.Codec }

That should be preferred given that it is already in place. I've attached a git diff for you to consider this approach (e.g. git apply yourcoworkers.diff).

kafka.diff.zip

Can we take this up in the next iteration of enhancement I can create a backlog item in our board for this. The reason I am asking is I need to update and add some test cases for this approach as we are dangerously close to release this might take more time :/

Sure no problem 👍

feat: initial draft to support initialisation of producer without topic

fcaf5a1

utsabc requested review from fracasula, atzoum and ItsSudip October 14, 2022 05:11

github-actions bot added server-team with tests labels Oct 14, 2022

utsabc requested review from koladilip and krishna2020 October 14, 2022 05:11

utsabc added 2 commits October 14, 2022 15:24

feat: added support to publish with topic info

9441ece

feat: added support to publish with topic for batched paylaods and mi…

d4bf244

…nor refactors

koladilip reviewed Oct 17, 2022

View reviewed changes

utsabc added 3 commits October 19, 2022 10:58

feat: addresed review comments

e9b0eca

feat: minor refactoring

5ea13e8

feat: updated client_test.go

c34af41

github-actions bot removed the with tests label Oct 19, 2022

atzoum reviewed Oct 20, 2022

View reviewed changes

$fracasula$

fracasula reviewed Oct 20, 2022

View reviewed changes

koladilip reviewed Oct 20, 2022

View reviewed changes

services/streammanager/kafka/client/producer.go Outdated Show resolved Hide resolved

utsabc added 3 commits October 24, 2022 18:58

feat(kafka multi topic): allowing topic to be passed only in produce …

3ae40f7

…removing client initialisation with topic

feat(kafka multi topic): refactoring function name

98b80bd

feat(kafka multi topic): removed vulnerablites from calling property …

cf618c7

…values from message

koladilip reviewed Oct 25, 2022

View reviewed changes

krishna2020 reviewed Oct 26, 2022

View reviewed changes

services/streammanager/kafka/kafkamanager.go Outdated Show resolved Hide resolved

feat(integrations): removed unwanted args from producer funcs

57f2257

koladilip reviewed Oct 26, 2022

View reviewed changes

services/streammanager/kafka/kafkamanager.go Outdated Show resolved Hide resolved

feat(kafka multi topic): refactored test cases for new implementation

e2d4428

github-actions bot added the with tests label Oct 26, 2022

utsabc added 5 commits November 5, 2022 12:38

feat: addressing comments on code refactoring

91abb1a

feat: addressing comments on code refactoring adding backward compati…

7fe1e6a

…bility support for default topic

feat: re-adding the sanity test cases

f66e8e9

Merge branch 'master' into feat.kafka-multi-topic-support

aa9d7e2

feat: adding default-topic-test case

560fa4e

atzoum reviewed Nov 7, 2022

View reviewed changes

services/streammanager/kafka/client/producer.go Show resolved Hide resolved

services/streammanager/kafka/kafkamanager.go Show resolved Hide resolved

feat: adding default-topic-test case for batch

4504a55

Merge branch 'master' into feat.kafka-multi-topic-support

9f7e642

utsabc changed the title ~~feat: initial draft to support initialisation of producer without topic~~ feat: support initialisation of producer without topic for kafka, azure event hub and confluent cloud Nov 7, 2022

$fracasula$

fracasula reviewed Nov 7, 2022

View reviewed changes

services/streammanager/kafka/kafkamanager_test.go Outdated Show resolved Hide resolved

$fracasula$

fracasula reviewed Nov 7, 2022

View reviewed changes

services/streammanager/kafka/kafkamanager_test.go Outdated Show resolved Hide resolved

$fracasula$

fracasula reviewed Nov 7, 2022

View reviewed changes

services/streammanager/kafka/kafkamanager_test.go Outdated Show resolved Hide resolved

$fracasula$

fracasula reviewed Nov 7, 2022

View reviewed changes

feat: clean up of test cases

eeb8667

atzoum approved these changes Nov 7, 2022

View reviewed changes

$fracasula$

fracasula approved these changes Nov 7, 2022

View reviewed changes

chandumlg merged commit 0312c55 into master Nov 7, 2022

chandumlg deleted the feat.kafka-multi-topic-support branch November 7, 2022 13:38

This was referenced Dec 1, 2022

chore: release 1.3.0 #2749

Closed

chore: prerelease 1.4.0-rc.1 #2750

Merged

chore: release 1.4.0 #2752

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: support initialisation of producer without topic for kafka, azure event hub and confluent cloud #2569

feat: support initialisation of producer without topic for kafka, azure event hub and confluent cloud #2569

utsabc commented Oct 14, 2022 •

edited

Loading

codecov bot commented Oct 14, 2022 •

edited

Loading

krishna2020 commented Oct 19, 2022

atzoum left a comment

atzoum Oct 20, 2022

utsabc Oct 24, 2022

atzoum Oct 20, 2022

utsabc Oct 24, 2022

atzoum Oct 20, 2022 •

edited

Loading

utsabc Oct 24, 2022

$@fracasula$ fracasula Nov 4, 2022

atzoum Oct 20, 2022

utsabc Oct 24, 2022

atzoum Nov 4, 2022

atzoum Oct 20, 2022

utsabc Oct 24, 2022

$@fracasula$ fracasula Oct 20, 2022

utsabc Oct 24, 2022

fracasula commented Oct 20, 2022

koladilip left a comment

atzoum commented Nov 7, 2022

utsabc commented Nov 7, 2022 •

edited

Loading

$@fracasula$ fracasula Nov 7, 2022

utsabc Nov 7, 2022

$@fracasula$ fracasula Nov 7, 2022

		@@ -132,6 +124,19 @@ func (p *Producer) Publish(ctx context.Context, msgs ...Message) error {

		var tempError interface{ Temporary() bool }

		func processHeaderForEachMessage(msg Message) (headers []kafka.Header) {

	func processHeaderForEachMessage(msg Message) (headers []kafka.Header) {
	func headers(msg Message) (headers []kafka.Header) {

	headers := processHeaderForEachMessage(msgs[i])
	headers := headers(msgs[i])

feat: support initialisation of producer without topic for kafka, azure event hub and confluent cloud #2569

feat: support initialisation of producer without topic for kafka, azure event hub and confluent cloud #2569

Conversation

utsabc commented Oct 14, 2022 • edited Loading

Description

Notion Ticket

Security

codecov bot commented Oct 14, 2022 • edited Loading

Codecov Report

krishna2020 commented Oct 19, 2022

atzoum 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

atzoum Oct 20, 2022 • edited Loading

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

Choose a reason for hiding this comment

fracasula commented Oct 20, 2022

koladilip left a comment

Choose a reason for hiding this comment

atzoum commented Nov 7, 2022

utsabc commented Nov 7, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

utsabc commented Oct 14, 2022 •

edited

Loading

codecov bot commented Oct 14, 2022 •

edited

Loading

atzoum Oct 20, 2022 •

edited

Loading

utsabc commented Nov 7, 2022 •

edited

Loading