Kafka 11 - record headers support #414

habutre · 2020-08-13T21:23:05Z

This PR is targeted to #397 to provide support for record headers provided since Kafka 0.11

The headers were implemented first on Kayrock and for testing purposes this branch points to Kayrock master. After a new Kayrock be released this PR can be changed and merged. This change is only for your review and validation.

To maintain compability while using the new client the Protocol.{Produce, Fetch}.Message had to be changed.

Only the new client is aware about such info on messages and may not cause conflicts or misbehavior

Out of scope: The mix format causes some changes that should be evaluate if make sense or have to be reverted

habutre · 2020-08-14T13:17:58Z

Hey @dantswain @jbruggem @joshuawscott @bjhaid I think you are the right guys to involve here to help me (let me know if someone might be added or removed from my mentions)

I am facing random errors with ci_tests.sh also with all_tests.sh locally, did you have some tips & tricks that I can apply to get rid of it and making the tests pass?

One can see in the CI tests each config has erros in different test cases and I am a bit lost to find a solution.

Thanks for any help.

joshuawscott · 2020-08-19T17:40:30Z

When I released 0.11.0 I fixed most of the intermittent failures, I thought. I was able to run 20x in a row without any failures locally. That said, because Travis uses 1.5 core machines, it doesn't necessarily work the same. There does seem to be a completely consistent failure, though:

  1) test gzip compression - produce v3, fetch v0 (KafkaEx.KayrockRecordBatchTest)
     test/integration/kayrock/record_batch_test.exs:301
     ** (UndefinedFunctionError) function nil.value/0 is undefined
     code: assert message.value == msg
     stacktrace:
       nil.value()
       test/integration/kayrock/record_batch_test.exs:328: (test)
     The following output was logged:
     
     21:27:44.845 [debug] Successfully connected to broker "localhost":9093
     
     21:27:44.888 [debug] Successfully connected to broker "localhost":9094
     
     21:27:44.932 [debug] Successfully connected to broker "localhost":9095

Often a single failure will cause ripples due to using the same underlying brokers, so my suggestion would be to solve this failure and see if the others begin to work.

habutre · 2020-09-11T20:02:28Z

Hey @dantswain I would really appreciate if you could take a look into my last commit that make the ci tests pass.
I am a bit affraid to change pre-existent tests to make all them pass.

@joshuawscott I think now you can review and if you can help to clarify about the offsets manipulation on kafka_ex/test/integration/kayrock/record_batch_test.exs will be awesome

ℹ️ Just a remind I am pointing the Kayrock to master and a release is necessary there before move forward with this PR

joshuawscott · 2020-11-02T14:38:59Z

kayrock 0.1.12 is out now

dantswain · 2020-11-02T14:47:26Z

@habutre I apologize for not getting to this yet. I have been unbelievably busy :( I will try to make time.

dantswain

@habutre I think this all looks good in theory, but it's hard to be sure of what I'm reviewing with all of the mix format changes included. Maybe after #420 is merged you could merge into your branch and update this PR?

lib/kafka_ex/protocol/fetch.ex

habutre · 2020-11-05T20:02:44Z

Is it possible to retry the CI? Checking the errors 1) timeout 2) failed to get deps

dantswain

I think we're good here. I kicked the build again and had one question about what seems like an unrelated change.

dantswain · 2020-11-05T21:13:23Z

test/integration/kayrock/record_batch_test.exs

@@ -234,7 +287,7 @@ defmodule KafkaEx.KayrockRecordBatchTest do

    fetch_responses =
      KafkaEx.fetch(topic, 0,
-        offset: max(offset - 2, 0),
+        offset: offset,


I don't remember exactly why this was set this way - probably some kind of bizarre test behavior. Did you have a compelling reason to change it?

At first I tried to understand the reason to get a previous offset instead the one returned by the producer, but since it always worked before I kept is AS-IS. After I have introduced the headers the kayrock tests start failing randomly but always with the same reason nil message.

I watch the topic with Kafka tool so I figure out that the offset - 2 doesn't match with the expected message then I tried to remove the max(offset - 2, 0) and got success on every execution what make me think that was the reason and also no other test failed.

I don't have a rational about that behavior only that on my mind the offset should not be manipulated to get the desired message, it was just a recursively try-n-check.

@dantswain in case you're interested, here is the commit where you added this max(offset - 2, 0): 7197a90

I didn't see anything in particular to help me understand. IMO the new changed code makes more sense 🤷

Argonus · 2021-05-31T08:55:05Z

@habutre @joshuawscott
Are we blocked by sth here? Maybe it's worth merging master to this PR & if everything is green merge it to the main branch.
If I can help somehow, please let me know.

habutre · 2021-05-31T15:14:03Z

@habutre @joshuawscott
Are we blocked by sth here? Maybe it's worth merging master to this PR & if everything is green merge it to the main branch.
If I can help somehow, please let me know.

Hey @Argonus this PR is ready for review and I am keeping as much as possible synced with latest merged PRs.

jbruggem

LGTM. I'd rather have a second opinion though :)

Argonus · 2021-10-29T13:23:34Z

@jbruggem @joshuawscott @dantswain
Can I help somehow with this PR?

To maintain compability while using the new client the message protocol had to be changed. Only the new client is aware about such info on messages

As usual I ran the command mix format that added changes to all these files. I don't know if run the `mix format` is a desired or common practice on _KafkaEx_ contribution, so I left it in a separated commit that can be reverted without big efforts

The offset resulting from the KafkaEx.produce sometimes was decreased causing some msg not being found or retrieved correct I don't have background to know why in the compressed msg testing the offset was subtract by -2 `max(offset-2, 0)` but the tests only passed when the retrieved offset was used to fetch messages

jbruggem

Still LGTM. @joshuawscott @dantswain could one of you cross-check these changes to make sure it makes sense ? I'd rather not approve alone.

lib/kafka_ex/protocol/produce.ex

dantswain

Yeesh sorry for the delay here. @habutre can you check @jbruggem 's comment, but otherwise looks good to me. Thanks for the work on this and for your patience!

lib/kafka_ex/protocol/produce.ex

jbruggem · 2021-11-02T20:13:16Z

@habutre do you wish to commit those last documentation suggestions ?

Once done, I'll merge.

Co-authored-by: Jehan Bruggeman <jbruggem@users.noreply.github.com>

Argonus · 2021-11-05T13:34:31Z

@jbruggem Maybe it's worth releasing a new version as well?
There would be two quite big things: support for snappy & support for headers

jbruggem · 2021-11-08T10:12:17Z

@Argonus TBH I've never cut a release on KafkaEx before, so I'm not sure what the criteria are :).

@dantswain @joshuawscott would this be a good moment to make a release, and if so who can do it ?

joshuawscott self-requested a review August 19, 2020 17:41

dantswain reviewed Nov 3, 2020

View reviewed changes

lib/kafka_ex/protocol/fetch.ex Outdated Show resolved Hide resolved

jbruggem mentioned this pull request Nov 3, 2020

Improve docker-compose by adding healthcheck condition at startup #420

Merged

dantswain reviewed Nov 5, 2020

View reviewed changes

habutre force-pushed the kafka-11-headers-support branch from 57a93c8 to 309e4c3 Compare January 15, 2021 21:39

habutre force-pushed the kafka-11-headers-support branch from 309e4c3 to 9b68ced Compare February 24, 2021 10:46

habutre force-pushed the kafka-11-headers-support branch from 9b68ced to 91e9626 Compare May 31, 2021 12:10

jbruggem approved these changes Jun 9, 2021

View reviewed changes

Rogério Ramos and others added 7 commits November 1, 2021 17:09

Make use of defined alias

1e75901

Add test case for record headers on new client

d641483

Add header key on produce and fetch message protocol

e725d88

To maintain compability while using the new client the message protocol had to be changed. Only the new client is aware about such info on messages

Add default timeout to produce request

a97ba8d

Specify headers type on fetch protocol

36fe923

habutre force-pushed the kafka-11-headers-support branch from 91e9626 to 36fe923 Compare November 1, 2021 16:10

jbruggem self-requested a review November 2, 2021 07:07

jbruggem approved these changes Nov 2, 2021

View reviewed changes

lib/kafka_ex/protocol/produce.ex Outdated Show resolved Hide resolved

dantswain approved these changes Nov 2, 2021

View reviewed changes

habutre commented Nov 2, 2021

View reviewed changes

lib/kafka_ex/protocol/produce.ex Show resolved Hide resolved

habutre and others added 2 commits November 2, 2021 21:20

Make producer headers type explicit

77e1b0d

Co-authored-by: Jehan Bruggeman <jbruggem@users.noreply.github.com>

Provide header param's doc

d285306

jbruggem merged commit eb21bb0 into kafkaex:master Nov 8, 2021

This was referenced May 9, 2022

record headers does not work with fetch v3 #465

Closed

Add kafka headers to messages, producer and consumer #397

Closed

jfmyers9 mentioned this pull request May 16, 2022

Create a new release with recent changes? #466

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Kafka 11 - record headers support #414

Kafka 11 - record headers support #414

habutre commented Aug 13, 2020

habutre commented Aug 14, 2020

joshuawscott commented Aug 19, 2020

habutre commented Sep 11, 2020 •

edited

Loading

joshuawscott commented Nov 2, 2020

dantswain commented Nov 2, 2020

dantswain left a comment

habutre commented Nov 5, 2020

dantswain left a comment

dantswain Nov 5, 2020

habutre Nov 6, 2020 •

edited

Loading

jbruggem Jun 9, 2021

Argonus commented May 31, 2021

habutre commented May 31, 2021

jbruggem left a comment

Argonus commented Oct 29, 2021

jbruggem left a comment

dantswain left a comment

jbruggem commented Nov 2, 2021

Argonus commented Nov 5, 2021

jbruggem commented Nov 8, 2021

Kafka 11 - record headers support #414

Kafka 11 - record headers support #414

Conversation

habutre commented Aug 13, 2020

habutre commented Aug 14, 2020

joshuawscott commented Aug 19, 2020

habutre commented Sep 11, 2020 • edited Loading

joshuawscott commented Nov 2, 2020

dantswain commented Nov 2, 2020

dantswain left a comment

Choose a reason for hiding this comment

habutre commented Nov 5, 2020

dantswain left a comment

Choose a reason for hiding this comment

dantswain Nov 5, 2020

Choose a reason for hiding this comment

habutre Nov 6, 2020 • edited Loading

Choose a reason for hiding this comment

jbruggem Jun 9, 2021

Choose a reason for hiding this comment

Argonus commented May 31, 2021

habutre commented May 31, 2021

jbruggem left a comment

Choose a reason for hiding this comment

Argonus commented Oct 29, 2021

jbruggem left a comment

Choose a reason for hiding this comment

dantswain left a comment

Choose a reason for hiding this comment

jbruggem commented Nov 2, 2021

Argonus commented Nov 5, 2021

jbruggem commented Nov 8, 2021

habutre commented Sep 11, 2020 •

edited

Loading

habutre Nov 6, 2020 •

edited

Loading