Add support for Kafka record headers #1025

bhavin192 · 2018-12-17T06:22:48Z

Add support for Kafka record headers
- This makes sure that the headers received from a function are
  converted to Kafka record headers
- Also extracts the headers from a Kafka record and sends them to
  function
- ref: https://issues.apache.org/jira/browse/KAFKA-4208
Make Kafka version configurable
- Adds version field to Kafka struct
- Adds MESSAGE_QUEUE_KAFKA_VERSION environment variable
- Drop headers from Kafka record if specified Kafka version is < 0.11.0.0
Update helm chart fission-all for Kafka record headers
- Adds new field kafka.version
- Sets the MESSAGE_QUEUE_KAFKA_VERSION environments variable to kafka.version

Update integration tests for Kafka headers support
- hellokafka.js: print header values and send the received headers as
  response
- kafka-pub.go: Add header Z-Custom-Name while producing the message
- test_kafka.sh:
  - Create another trigger which calls consumer-func2; consumer-func2
    will print the headers it receives
  - Producer -(testtopic)-> MQT --> consumer-func --> MQT -(resptopic)-> MQT --> consumer-func2
  - Use timeout instead of gtimout (available only on Mac OSX)
  - do glide install before creating package zip file
  - New function test_fnmessage which checks for a message in a function's log

This change is

bhavin192 · 2018-12-19T09:59:41Z

Not sure why the build is failing. It passes on my fork which has same .travis.yaml (only allowed branches and go_import_path: github.com/fission/fission are changed)
https://travis-ci.org/bhavin192/fission/jobs/469912067/config

life1347

Reviewed 1 of 4 files at r1.
Reviewable status: 1 of 4 files reviewed, 2 unresolved discussions (waiting on @bhavin192 and @life1347)

mqtrigger/messageQueue/kafka.go, line 32 at r1 (raw file):

	"github.com/fission/fission"
	"github.com/fission/fission/crd"
	"os"

please sort the import

mqtrigger/messageQueue/kafka.go, line 52 at r1 (raw file):

	kafkaVersion, err := sarama.ParseKafkaVersion(mqKafkaVersion)
	if err != nil {
		log.Warningf("error parsing version string %q: %v. Falling back to %q", mqKafkaVersion, err, kafkaVersion)

nitpick: Error parsing
(uppercase)

life1347 · 2018-12-27T07:06:40Z

LGTM, just two minor comments. BTW, the PR description for changes is awesome.

bhavin192 · 2018-12-27T08:19:31Z

Hey @life1347 I have made changes according to your suggestions.

BTW, the PR description for changes is awesome.

Hehe, thanks :)

bhavin192

Reviewable status: 1 of 4 files reviewed, 2 unresolved discussions (waiting on @life1347)

mqtrigger/messageQueue/kafka.go, line 32 at r1 (raw file):

Previously, life1347 (Ta-Ching Chen) wrote…

please sort the import

Done.

mqtrigger/messageQueue/kafka.go, line 52 at r1 (raw file):

Previously, life1347 (Ta-Ching Chen) wrote…

nitpick: Error parsing
(uppercase)

Ah! right

life1347 · 2019-01-09T06:16:59Z

LGTM. please rebase master and we can merge it when ci passed

- This makes sure that the headers received from a function are converted to Kafka record headers - Also extracts the headers from a Kafka record and sends them to function - ref: https://issues.apache.org/jira/browse/KAFKA-4208 Signed-off-by: Bhavin Gandhi <bhavin7392@gmail.com>

- Adds version field to Kafka struct - Adds MESSAGE_QUEUE_KAFKA_VERSION environment variable - Drop the headers if Kafka version is < 0.11.0.0 Signed-off-by: Bhavin Gandhi <bhavin7392@gmail.com>

- Adds new field `kafka.version` - Sets the MESSAGE_QUEUE_KAFKA_VERSION environments variable to `kafka.version` Signed-off-by: Bhavin Gandhi <bhavin7392@gmail.com>

bhavin192 · 2019-01-11T10:52:57Z

Here is the output of integration test of kafka

Creating node-kafka environment
environment 'node-kafka' created
Creating go-kafka environment
environment 'go-kafka' created
Creating package for Kafka producer
...
Output of glide install
...
pkgName=kafka-zip-3lwb
~/work/go/src/github.com/fission/fission/test/tests/mqtrigger/kafka
Waiting for builder manager to finish the build
Waiting for build to finish
Waiting for build to finish
Waiting for build to finish
Waiting for build to finish
succeeded
Package kafka-zip-3lwb created
Creating function consumer-func
function 'consumer-func' created
Creating function consumer-func2
function 'consumer-func2' created
Creating function producer-func
function 'producer-func' created
Creating trigger kafkatest
trigger 'kafkatest' created
Creating trigger kafkatest2
trigger 'kafkatest2' created
Successfully sent to testtopicTesting pool manager function
Checking for valid response
level=info msg="Calling message handler with value {"name": "testvalue"}" time="2019-01-11T10:07:18Z" level=info msg="Making HTTP request to http://router.fission/fission-function/consumer-func"
Testing the headers values in consumer-func
Checking for valid function log
> fission-nodejs-runtime@0.1.0 start /usr/src/app > node server.js user code loaded in 0sec 0.354236ms ::1 - - [11/Jan/2019:10:03:23 +0000] "POST /specialize HTTP/1.1" 202 - "-" "Go-http-client/1.1" { name: 'testvalue' } z-custom-name: Kafka-Header-test x-fission-function-name: consumer-func ::ffff:10.36.0.6 - - [11/Jan/2019:10:03:23 +0000] "POST / HTTP/1.1" 200 20 "-" "Go-http-client/1.1" > fission-nodejs-runtime@0.1.0 start /usr/src/app > node server.js user code loaded in 0sec 3.618165ms ::1 - - [11/Jan/2019:10:07:18 +0000] "POST /specialize HTTP/1.1" 202 - "-" "Go-http-client/1.1" { name: 'testvalue' } z-custom-name: Kafka-Header-test x-fission-function-name: consumer-func ::ffff:10.36.0.6 - - [11/Jan/2019:10:07:18 +0000] "POST / HTTP/1.1" 200 20 "-" "Go-http-client/1.1"
Testing the header value in consumer-func2
Checking for valid function log
> fission-nodejs-runtime@0.1.0 start /usr/src/app > node server.js user code loaded in 0sec 0.222792ms ::1 - - [11/Jan/2019:10:03:23 +0000] "POST /specialize HTTP/1.1" 202 - "-" "Go-http-client/1.1" { name: 'testvalue' } z-custom-name: Kafka-Header-test x-fission-function-name: consumer-func2 ::ffff:10.36.0.6 - - [11/Jan/2019:10:03:23 +0000] "POST / HTTP/1.1" 200 20 "-" "Go-http-client/1.1" > fission-nodejs-runtime@0.1.0 start /usr/src/app > node server.js user code loaded in 0sec 0.407005ms ::1 - - [11/Jan/2019:10:07:19 +0000] "POST /specialize HTTP/1.1" 202 - "-" "Go-http-client/1.1" { name: 'testvalue' } z-custom-name: Kafka-Header-test x-fission-function-name: consumer-func2 ::ffff:10.36.0.6 - - [11/Jan/2019:10:07:19 +0000] "POST / HTTP/1.1" 200 20 "-" "Go-http-client/1.1"
Checking for valid function log
> fission-nodejs-runtime@0.1.0 start /usr/src/app > node server.js user code loaded in 0sec 0.222792ms ::1 - - [11/Jan/2019:10:03:23 +0000] "POST /specialize HTTP/1.1" 202 - "-" "Go-http-client/1.1" { name: 'testvalue' } z-custom-name: Kafka-Header-test x-fission-function-name: consumer-func2 ::ffff:10.36.0.6 - - [11/Jan/2019:10:03:23 +0000] "POST / HTTP/1.1" 200 20 "-" "Go-http-client/1.1" > fission-nodejs-runtime@0.1.0 start /usr/src/app > node server.js user code loaded in 0sec 0.407005ms ::1 - - [11/Jan/2019:10:07:19 +0000] "POST /specialize HTTP/1.1" 202 - "-" "Go-http-client/1.1" { name: 'testvalue' } z-custom-name: Kafka-Header-test x-fission-function-name: consumer-func2 ::ffff:10.36.0.6 - - [11/Jan/2019:10:07:19 +0000] "POST / HTTP/1.1" 200 20 "-" "Go-http-client/1.1"
Cleaning up...
environment 'go-kafka' deleted
environment 'node-kafka' deleted
function 'producer-func' deleted
function 'consumer-func' deleted
function 'consumer-func2' deleted
trigger 'kafkatest' deleted
trigger 'kafkatest2' deleted

- hellokafka.js: print header values and send the received headers as response - kafka-pub.go: Add header Z-Custom-Name while producing the message - test_kafka.sh: - Create another trigger which calls consumer-func2; consumer-func2 will print the headers it receives - Producer -(testtopic)-> MQT --> consumer-func --> MQT -(resptopic)-> MQT --> consumer-func2 - Use timeout instead of gtimout (available only on Mac OSX) - do glide install before creating package zip file - New function test_fnmessage which checks for a message in a function's log Signed-off-by: Bhavin Gandhi <bhavin7392@gmail.com>

vishal-biyani · 2019-01-11T14:31:08Z

@life1347 The build fails with same PR's code here: #1052 and also the test was enhanced to test changes in this PR and tested locally (As it needs Kafka cluster setup). Looks good?

life1347 · 2019-01-11T14:36:13Z

yes, looks great. We can merge it 🎉

life1347

Reviewed 1 of 4 files at r1, 2 of 2 files at r2, 3 of 3 files at r3.
Reviewable status: complete! all files reviewed, all discussions resolved

vishal-biyani · 2019-01-12T15:43:17Z

Build in #1052 is green for the same change. This change has been tested in a local setup with Kafka cluster and working as expected with test. Merging.

Thanks a lot, @bhavin192 for your contribution!

bhavin192 changed the title ~~Add support for Kafka record headers~~ [WIP] Add support for Kafka record headers Dec 17, 2018

bhavin192 force-pushed the kafka-trigger-headers branch from 13c4584 to b01041d Compare December 18, 2018 13:14

bhavin192 changed the title ~~[WIP] Add support for Kafka record headers~~ Add support for Kafka record headers Dec 19, 2018

bhavin192 force-pushed the kafka-trigger-headers branch 2 times, most recently from 14d1b35 to 8ad4e03 Compare December 20, 2018 18:46

vishal-biyani added this to the 1.0-rc2 milestone Dec 26, 2018

vishal-biyani requested a review from life1347 December 26, 2018 09:03

bhavin192 force-pushed the kafka-trigger-headers branch from 8ad4e03 to 57be599 Compare December 26, 2018 09:03

life1347 requested changes Dec 27, 2018

View reviewed changes

bhavin192 force-pushed the kafka-trigger-headers branch from 57be599 to fe1d974 Compare December 27, 2018 07:28

bhavin192 force-pushed the kafka-trigger-headers branch from fe1d974 to f3ee39c Compare December 27, 2018 08:36

vishal-biyani mentioned this pull request Dec 27, 2018

Kafka Headers in documentation fission/docs.fission.io#102

Closed

bhavin192 commented Jan 9, 2019

View reviewed changes

bhavin192 force-pushed the kafka-trigger-headers branch from f3ee39c to 31db92f Compare January 9, 2019 06:40

vishal-biyani mentioned this pull request Jan 11, 2019

Kafka header change #1052

Closed

bhavin192 added 3 commits January 11, 2019 15:48

Make Kafka version configurable

5103cba

- Adds version field to Kafka struct - Adds MESSAGE_QUEUE_KAFKA_VERSION environment variable - Drop the headers if Kafka version is < 0.11.0.0 Signed-off-by: Bhavin Gandhi <bhavin7392@gmail.com>

Update helm chart fission-all for Kafka record headers

9ce6a4c

- Adds new field `kafka.version` - Sets the MESSAGE_QUEUE_KAFKA_VERSION environments variable to `kafka.version` Signed-off-by: Bhavin Gandhi <bhavin7392@gmail.com>

bhavin192 force-pushed the kafka-trigger-headers branch from 31db92f to 18978bd Compare January 11, 2019 10:20

bhavin192 force-pushed the kafka-trigger-headers branch from 18978bd to f3f0413 Compare January 11, 2019 13:30

life1347 approved these changes Jan 11, 2019

View reviewed changes

life1347 approved these changes Jan 12, 2019

View reviewed changes

vishal-biyani merged commit 37f37c0 into fission:master Jan 12, 2019

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add support for Kafka record headers #1025

Add support for Kafka record headers #1025

bhavin192 commented Dec 17, 2018 •

edited

bhavin192 commented Dec 19, 2018

life1347 left a comment

life1347 commented Dec 27, 2018 •

edited

bhavin192 commented Dec 27, 2018

bhavin192 left a comment

life1347 commented Jan 9, 2019

bhavin192 commented Jan 11, 2019

vishal-biyani commented Jan 11, 2019

life1347 commented Jan 11, 2019

life1347 left a comment

vishal-biyani commented Jan 12, 2019

Add support for Kafka record headers #1025

Add support for Kafka record headers #1025

Conversation

bhavin192 commented Dec 17, 2018 • edited

bhavin192 commented Dec 19, 2018

life1347 left a comment

Choose a reason for hiding this comment

life1347 commented Dec 27, 2018 • edited

bhavin192 commented Dec 27, 2018

bhavin192 left a comment

Choose a reason for hiding this comment

life1347 commented Jan 9, 2019

bhavin192 commented Jan 11, 2019

vishal-biyani commented Jan 11, 2019

life1347 commented Jan 11, 2019

life1347 left a comment

Choose a reason for hiding this comment

vishal-biyani commented Jan 12, 2019

bhavin192 commented Dec 17, 2018 •

edited

life1347 commented Dec 27, 2018 •

edited