Fix Rebalance events behavior for static membership and fix for consu… #757

jliunyu · 2022-04-06T01:02:55Z

…mer close takes longer than 10s

EAGER rebalance: 6 partition(s) revoked: [testrebalance[0]@unset testrebalance[1]@unset testrebalance[2]@unset testrebalance[3]@unset testrebalance[4]@unset testrebalance[5]@unset]
Ignored OffsetsCommitted (<nil>, [testrebalance[0]@953 testrebalance[1]@900 testrebalance[2]@919 testrebalance[3]@unset testrebalance[4]@857 testrebalance[5]@925]) for consumer rdkafka#consumer-1
rdkafka#consumer-1, EAGER rebalance: 3 new partition(s) assigned: [testrebalance[0]@unset testrebalance[1]@unset testrebalance[2]@unset]
rdkafka#consumer-2, EAGER rebalance: 3 new partition(s) assigned: [testrebalance[3]@unset testrebalance[4]@unset testrebalance[5]@unset]
Message on testrebalance[1]@900:
test
EAGER rebalance: 3 partition(s) revoked: [testrebalance[0]@unset testrebalance[1]@unset testrebalance[2]@unset]
Ignored OffsetsCommitted (<nil>, [testrebalance[0]@unset testrebalance[1]@901 testrebalance[2]@unset]) for consumer rdkafka#consumer-1
EAGER rebalance: 3 partition(s) revoked: [testrebalance[3]@unset testrebalance[4]@unset testrebalance[5]@unset]
rdkafka#consumer-2, EAGER rebalance: 2 new partition(s) assigned: [testrebalance[2]@unset testrebalance[3]@unset]
rdkafka#consumer-1, EAGER rebalance: 2 new partition(s) assigned: [testrebalance[0]@unset testrebalance[1]@unset]
rdkafka#consumer-3, EAGER rebalance: 2 new partition(s) assigned: [testrebalance[4]@unset testrebalance[5]@unset]
rdkafka#consumer-4, EAGER rebalance: 2 new partition(s) assigned: [testrebalance[2]@unset testrebalance[3]@unset]
PASS
ok  	_/Users/jliu/Documents/Code/jliunyu/confluent-kafka-go/kafka	38.019s

Manually tested for the Close, before this fix, the consumer takes more than 10s, with this fix, closing time: 730.602µs

consumer closed. closing time: 730.602µs
PASS
ok  	_/Users/jliu/Documents/Code/jliunyu/confluent-kafka-go/kafka	0.573s

CLAassistant · 2022-04-06T01:03:00Z

All committers have signed the CLA.

edenhill

Good stuff!

CHANGELOG.md

edenhill · 2022-04-06T09:02:45Z

kafka/consumer.go

+	for run {
+		c.Poll(100)
+		select {
+		case <-doneChan:


I think this will block until doneChan is written to, which means if Poll() is triggered once for some other reason, e.g., a stats callback, then it will hang here.

Can we use int(C.rd_kafka_queue_length(c.handle.rkq)) == 0 to exist the loop which is same with the previous way?

for run { c.Poll(100) select { case <-doneChan: run = false default: if int(C.rd_kafka_queue_length(c.handle.rkq)) == 0 { run = false } } }

And I'm also confused why it will hang, I tested with this code with the existing change in this PR, it doesn't hang

func TestConsumerClose1(t *testing.T) { c, err := NewConsumer(&ConfigMap{ "group.id": "gotest"}) if err != nil { t.Fatalf("%s", err) } startClose := time.Now() err = c.Close() if err != nil { panic(err) } fmt.Println(fmt.Sprintf("consumer closed. closing time: %v", time.Since(startClose))) }

I think with the following sequence of events it will hang:

application calls Close()

Close() is blocking on c.Poll()

A stats event (or something else which is not a rebalance event) triggers the Poll and Poll() wakes up and returns

Close() is now blocking on doneChan

librdkafka emits a rebalance event and waits for the app/go to ack it

.. but since Close() is no longer calling Poll() the rebalance event is never handled.

indefinite hang

I think this is easily solved by moving the Poll to the default case in the select, like in other places in the code.

for run { select { case <-doneChan: **-------------->// Is it possible that after the doneChan, there are still other events we need to handle? We always have to wait to read the doneChan value. If we exit from default, the doneChan will report some error** run = false default: c.Poll(100) if int(C.rd_kafka_queue_length(c.handle.rkq)) == 0 { run = false } } }

I don't think so, when doneChan is done we don't need to call Poll no more.

Why is the queue_length() check needed?

We don't need it, I already removed it from my code.

kafka/consumer_test.go

edenhill · 2022-04-06T09:07:50Z

kafka/consumer_test.go

+	switch ev := event.(type) {
+	case AssignedPartitions:
+		atomic.AddInt32(&assignedEvents, 1)
+		fmt.Fprintf(os.Stderr,


nit: Tests should use t.Logf if possible, but that might be tricky here in the callback.

Yeah, is there any good way to use t *testing.T in the callback?

I don't know, really. Go 🤷

kafka/consumer_test.go

edenhill

Do all existing tests pass now?

CHANGELOG.md

edenhill · 2022-04-14T06:56:52Z

kafka/consumer_test.go

@@ -429,3 +431,277 @@ func TestConsumerLog(t *testing.T) {
 		}
 	}
 }
+
+func wrapRebalanceCb(assignedEvents *int32, revokedEvents *int32, t *testing.T) func(c *Consumer, event Event) error {


👍 on the wrapping to provide the testing.T

kafka/consumer_test.go

edenhill · 2022-04-14T07:00:40Z

kafka/consumer_test.go

+		return
+	}
+
+	wg := &sync.WaitGroup{}


I think it is better to keep this as a non-pointer here, and then pass it as pointer when needed (to testPoll()).

If we keep this as a non-pointer here, we need to use it as pointer each time, and there are a lot of use cases in the same function.

For example:
wg.Add(1)
wg.Wait()

The pointer receiver methods apply also to non pointer variables, based on this.

So pointer should be needed to pass the WaitGroup struct by reference, but not to call its methods with pointer receiver.

If we keep this as a non-pointer here, we need to use it as pointer each time, and there are a lot of use cases in the same function.

I still believe it is more correct to keep it as a non-pointer here and pass it as a pointer where needed.
But it is a nit, so do as you wish.

jliunyu · 2022-04-15T05:16:24Z

Do all existing tests pass now?

Yes:)

jliu@C02DM32YMD6T kafka % go test -run "TestConsumer*" 
%4|1649999288.094|CONFWARN|rdkafka#consumer-1| [thrd:app]: Configuration property `fetch.wait.max.ms` (500) should be set lower than `socket.timeout.ms` (1000) by at least 1000ms to avoid blocking and timing out sub-sequent requests
rdkafka#consumer-6 Assigned partitions are: [testrebalance[0]@stored testrebalance[1]@stored testrebalance[2]@stored testrebalance[3]@stored testrebalance[4]@stored testrebalance[5]@stored]
rdkafka#consumer-7 Assigned partitions are: [testrebalance[3]@stored testrebalance[4]@stored testrebalance[5]@stored]
rdkafka#consumer-8 Assigned partitions are: [testrebalance[4]@stored testrebalance[5]@stored]
rdkafka#consumer-9 Assigned partitions are: [testrebalance[2]@stored testrebalance[3]@stored]
Static test passed

…mer close takes longer than 10s

jliunyu · 2022-04-19T16:05:23Z

This is the CI test: https://app.travis-ci.com/github/confluentinc/confluent-kafka-go/builds/249581975

kafka/consumer_test.go

edenhill

Still some unaddressed comments in the test file.

edenhill · 2022-04-28T12:36:44Z

kafka/consumer_test.go

+		return
+	}
+
+	wg := &sync.WaitGroup{}


If we keep this as a non-pointer here, we need to use it as pointer each time, and there are a lot of use cases in the same function.

I still believe it is more correct to keep it as a non-pointer here and pass it as a pointer where needed.
But it is a nit, so do as you wish.

kafka/consumer_test.go

edenhill

LGTM!

sermojohn · 2022-05-20T09:08:50Z

Is there a plan to make a release with this fix?

jliunyu · 2022-05-20T18:00:02Z

Is there a plan to make a release with this fix?

Yes, this one will be released soon.

confluentinc#757)

jliunyu requested review from edenhill and mhowlett April 6, 2022 01:23

jliunyu mentioned this pull request Apr 6, 2022

Closing consumer takes 10 second #661

Closed

7 tasks

edenhill suggested changes Apr 6, 2022

View reviewed changes

edenhill suggested changes Apr 14, 2022

View reviewed changes

Fix Rebalance events behavior for static membership and fix for consu…

f9ff23f

…mer close takes longer than 10s

jliunyu force-pushed the staticMember branch 2 times, most recently from fe62e71 to f9ff23f Compare April 19, 2022 16:03

jliunyu mentioned this pull request Apr 21, 2022

Consumer hanging on Close() call when redeploying application #767

Closed

sermojohn reviewed Apr 28, 2022

View reviewed changes

kafka/consumer_test.go Outdated Show resolved Hide resolved

sermojohn reviewed Apr 28, 2022

View reviewed changes

kafka/consumer_test.go Outdated Show resolved Hide resolved

edenhill suggested changes Apr 28, 2022

View reviewed changes

Handle review comments

082c093

jliunyu force-pushed the staticMember branch from 5efeabd to 082c093 Compare April 29, 2022 05:31

edenhill approved these changes May 2, 2022

View reviewed changes

jliunyu merged commit e01dd29 into confluentinc:master May 2, 2022

emasab pushed a commit to emasab/confluent-kafka-go that referenced this pull request May 26, 2022

Fix Rebalance events behavior for static membership and fix for consu… (

a521d28

confluentinc#757)

jaime0815 mentioned this pull request Jul 20, 2022

Upgrade confluent kafka go version milvus-io/milvus#18293

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fix Rebalance events behavior for static membership and fix for consu… #757

Fix Rebalance events behavior for static membership and fix for consu… #757

jliunyu commented Apr 6, 2022 •

edited

CLAassistant commented Apr 6, 2022 •

edited

edenhill left a comment

edenhill Apr 6, 2022

jliunyu Apr 7, 2022

jliunyu Apr 7, 2022

edenhill Apr 7, 2022

edenhill Apr 7, 2022

jliunyu Apr 7, 2022

edenhill Apr 7, 2022

edenhill Apr 7, 2022

jliunyu Apr 7, 2022

edenhill Apr 6, 2022

jliunyu Apr 7, 2022

edenhill Apr 7, 2022

edenhill left a comment

edenhill Apr 14, 2022

edenhill Apr 14, 2022

jliunyu Apr 15, 2022

sermojohn Apr 28, 2022

edenhill Apr 28, 2022

jliunyu commented Apr 15, 2022

jliunyu commented Apr 19, 2022

edenhill left a comment

edenhill Apr 28, 2022

edenhill left a comment

sermojohn commented May 20, 2022

jliunyu commented May 20, 2022

Fix Rebalance events behavior for static membership and fix for consu… #757

Fix Rebalance events behavior for static membership and fix for consu… #757

Conversation

jliunyu commented Apr 6, 2022 • edited

CLAassistant commented Apr 6, 2022 • edited

edenhill 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

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

edenhill 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

jliunyu commented Apr 15, 2022

jliunyu commented Apr 19, 2022

edenhill left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

edenhill left a comment

Choose a reason for hiding this comment

sermojohn commented May 20, 2022

jliunyu commented May 20, 2022

jliunyu commented Apr 6, 2022 •

edited

CLAassistant commented Apr 6, 2022 •

edited