consume speed control and rebalance questions

[imqishi]

Versions

Please specify real version numbers or git SHAs, not just "Latest" since that changes fairly regularly.

Sarama	Kafka	Go
1.29.1	1.1.1	1.17.2

Configuration

What configuration values are you using for Sarama and Kafka?

cfg.Consumer.Offsets.Initial = sarama.OffsetNewest
cfg.Consumer.Return.Errors = false
cfg.Consumer.Offsets.AutoCommit.Interval = 3 * time.Second
cfg.Consumer.MaxWaitTime = time.Second
cfg.Consumer.Fetch.Default = 524288
cfg.Consumer.Fetch.Max = 1048576
cfg.Consumer.Group.Rebalance.Timeout = 10 * time.Second
cfg.Net.MaxOpenRequests = 1
cfg.Net.KeepAlive = 1 * time.Minute
cfg.Metadata.Retry.Max = 1
cfg.Metadata.Retry.Backoff = 1000 * time.Millisecond
cfg.Metadata.RefreshFrequency = 5 * time.Minute
cfg.Metadata.Full = false

Logs

When filing an issue please provide logs from Sarama and Kafka if at all
possible. You can set sarama.Logger to a log.Logger to capture Sarama debug
output.

logs: CLICK ME

well, if consume one message cost too long (such as 3s), there will be some errors like "consumer is not correct in current generation"

in this situation message will be consumed repeated, and consume state is in rebalance

a simple way to recur this is write a simple consume code and in consume chan loop to sleep 3 second

Problem Description

I try to debug sarama, print some logs and learn some about kafka then deduce that consume time too long cause client and broker disconnected from each other so that generation is timeout, but what I cannot understand is I see there will be a goroutine to send loop heartbeat to broker to keep alive, it is dependent with consumer goroutine, why is client still disconnected? Is it because I only use sess.MarkMessage in consume loop but msg commit is in another goroutine, too long time there's no commit invoked? Besides reducing processing latency, is there any other way to handle this?

The second problem is that we want to control consume speed such as use a rate limiter for each partition, but I'm worried that the first problem will occur if rate limiter cause consume slow.

[github-actions]

Thank you for taking the time to raise this issue. However, it has not had any activity on it in the past 90 days and will be closed in 30 days if no updates occur.
Please check if the main branch has already resolved the issue since it was raised. If you believe the issue is still valid and you would like input from the maintainers then please comment to ask for it to be reviewed.

[dnwe]

@imqishi Sarama has MaxProcessingTime as a config parameter similar to the Java client's max.poll.interval.ms

sarama/config.go

Lines 391 to 408 in d2dba29

	// The maximum amount of time the consumer expects a message takes to
	// process for the user. If writing to the Messages channel takes longer
	// than this, that partition will stop fetching more messages until it
	// can proceed again.
	// Note that, since the Messages channel is buffered, the actual grace time is
	// (MaxProcessingTime * ChannelBufferSize). Defaults to 100ms.
	// If a message is not written to the Messages channel between two ticks
	// of the expiryTicker then a timeout is detected.
	// Using a ticker instead of a timer to detect timeouts should typically
	// result in many fewer calls to Timer functions which may result in a
	// significant performance improvement if many messages are being sent
	// and timeouts are infrequent.
	// The disadvantage of using a ticker instead of a timer is that
	// timeouts will be less accurate. That is, the effective timeout could
	// be between `MaxProcessingTime` and `2 * MaxProcessingTime`. For
	// example, if `MaxProcessingTime` is 100ms then a delay of 180ms
	// between two messages being sent may not be recognized as a timeout.
	MaxProcessingTime time.Duration

That is what you are seeing with the "consuming was taking too long" output from Sarama, it is this locally enforced timeout being hit and is separate from the heartbeat to the server which keeps the session alive. See "Decoupling the processing timeout" in KIP-62

[imqishi]

Thanks for your information which helps me understand the inner logic deeper, but after my trying the weird thing is that it seems

cfg.Consumer.Group.Rebalance.Timeout = 10 * time.Second

this config causes the timeout, in my team someone set it to 10 seconds, but the default config is 60 seconds. After reset it the consuming is correct

[dnwe]

So the rebalance timeout is what the consumer sends in its JoinGroup request and sets the maximum amount of time that the coordinator should wait for each member to rejoin when rebalancing the group. So that suggests that rebalancing had been triggered whilst the consumer was busy doing something and it didn't notice and re-join in time before the rebalance timeout was exceeded

[github-actions]

Thank you for taking the time to raise this issue. However, it has not had any activity on it in the past 90 days and will be closed in 30 days if no updates occur.
Please check if the main branch has already resolved the issue since it was raised. If you believe the issue is still valid and you would like input from the maintainers then please comment to ask for it to be reviewed.