Unexpected user-specified time limit error still occurs

[kimtree]

Versions

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

Sarama	Kafka	Go
v1.26.4	v1.1.1	v1.13.1

Configuration

What configuration values are you using for Sarama and Kafka?

config := sarama.NewConfig()
config.Version = sarama.V1_1_1_0
config.Consumer.Return.Errors = true

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

time="2019-12-13 15:38:23.746" level=info msg="client/metadata fetching metadata for [kafka_topic_name] from broker BROKER_SERVER3.local:9092"
time="2019-12-13 15:38:23.746" level=info msg="client/metadata fetching metadata for [kafka_topic_name] from broker BROKER_SERVER2.local:9092"
time="2019-12-13 15:38:23.890" level=info msg="client/coordinator requesting coordinator for consumergroup consumer_group_name from BROKER_SERVER4.local:9092"
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/9: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/11: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/10: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/67: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.890" level=info msg="client/coordinator requesting coordinator for consumergroup consumer_group_name from BROKER_SERVER3.local:9092"
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/12: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/7: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/68: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/8: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/6: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/65: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/50: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/54: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/66: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/55: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/63: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/64: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=info msg="client/coordinator coordinator for consumergroup consumer_group_name is #3 (BROKER_SERVER3.local:9092)"
time="2019-12-13 15:38:23.891" level=info msg="client/coordinator requesting coordinator for consumergroup consumer_group_name from BROKER_SERVER2.local:9092"
time="2019-12-13 15:38:23.891" level=info msg="client/coordinator requesting coordinator for consumergroup consumer_group_name from BROKER_SERVER2.local:9092"
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/51: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/52: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/53: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/1: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/4: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/2: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/3: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/5: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=error msg="kafka: error while consuming kafka_topic_name/0: kafka server: Request exceeded the user-specified time limit in the request." logLevel=ERROR
time="2019-12-13 15:38:23.891" level=info msg="client/coordinator coordinator for consumergroup consumer_group_name is #3 (BROKER_SERVER3.local:9092)"
time="2019-12-13 15:38:23.891" level=info msg="client/coordinator coordinator for consumergroup consumer_group_name is #3 (BROKER_SERVER3.local:9092)"
time="2019-12-13 15:38:23.892" level=info msg="client/coordinator coordinator for consumergroup consumer_group_name is #3 (BROKER_SERVER3.local:9092)"
time="2019-12-13 15:38:26.726" level=info msg="client/metadata fetching metadata for [kafka_topic_name] from broker BROKER_SERVER4.local:9092"
time="2019-12-13 15:38:26.726" level=info msg="client/metadata fetching metadata for [kafka_topic_name] from broker BROKER_SERVER3.local:9092"
time="2019-12-13 15:38:26.726" level=info msg="client/metadata fetching metadata for [kafka_topic_name] from broker BROKER_SERVER2.local:9092"
time="2019-12-13 15:38:26.725" level=info msg="client/metadata fetching metadata for [kafka_topic_name] from broker BROKER_SERVER2.local:9092"
time="2019-12-13 15:38:26.891" level=info msg="client/metadata fetching metadata for [kafka_topic_name] from broker BROKER_SERVER2.local:9092"
time="2019-12-13 15:38:26.891" level=info msg="client/metadata fetching metadata for [kafka_topic_name] from broker BROKER_SERVER2.local:9092"
time="2019-12-13 15:38:26.891" level=info msg="client/metadata fetching metadata for [kafka_topic_name] from broker BROKER_SERVER3.local:9092"

Problem Description

This was the original issue (#1562)
After I update Sarama version to v1.26.4 and cleaned config values but, I still got unexpected user-specified time limit errors from the broker while consuming the topic. Is it related to the Broker issue? or Can we do anything from the client-side?

Below is the code where the error comes from.

go func() {
	for err := range group.Errors() {
		log.Error(err)
	}
}()

[qiuxuxu1]

I am experiencing the the same problem.
Is it safe to just ignore this error?

[kimtree]

@qiuxuxu1 To me, It seems no problem while consuming the data (looks like this request will be retried).

[fmanna-c4w]

I have the same problem with sarama v1.17.2.
When error occurs, i can see a peak of the lag, so something happens, but it's not trigger rebalance, retry BackoffFunc or other.

[ManiSanganna]

I am having same issue.. error logs are printing. in my case I set auto commit to false
consumer:
auto commit to false

[un000]

The same, no rebalances, just errors kafka server: Request exceeded the user-specified time limit in the request.

[ChengJoyceJi]

Experiencing the same with Sarama v1.29.1 (current latest).
My config is:

config.Consumer.Group.Rebalance.Timeout = 60s
config.Consumer.Group.Session.Timeout = 50s
config.Consumer.Group.Heartbeat.Interval = 5s
config.Consumer.Return.Errors = true
config.Consumer.MaxProcessingTime = 10s

[dnwe]

We need to resurrect #2040 and see if that solves the OP's issues

[mingdyuo]

@dnwe
Thank you for taking care of this issue, and I'm facing same issue.

kafka: error while consuming topic_name/partition_number: kafka server: Request exceeded the user-specified time limit in the request

Is it safe to ignore this error? And does this timeout error have no effect on consume result due to request retry?

[dnwe]

@mingdyuo for that error you should either increase your config.Consumer.MaxProcessingTime config option or modify your ConsumeClaim loop so it isn't blocking for too long processing messages. Start by making sure it looks similar to the example

sarama/examples/consumergroup/main.go

Lines 180 to 204 in 9b0419d

	// ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().
	// Once the Messages() channel is closed, the Handler must finish its processing
	// loop and exit.
	func (consumer *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	// NOTE:
	// Do not move the code below to a goroutine.
	// The `ConsumeClaim` itself is called within a goroutine, see:
	// https://github.com/IBM/sarama/blob/main/consumer_group.go#L27-L29
	for {
	select {
	case message, ok := <-claim.Messages():
	if !ok {
	log.Printf("message channel was closed")
	return nil
	}
	log.Printf("Message claimed: value = %s, timestamp = %v, topic = %s", string(message.Value), message.Timestamp, message.Topic)
	session.MarkMessage(message, "")
	// Should return when `session.Context()` is done.
	// If not, will raise `ErrRebalanceInProgress` or `read tcp <ip>:<port>: i/o timeout` when kafka rebalance. see:
	// https://github.com/IBM/sarama/issues/1192
	case <-session.Context().Done():
	return nil
	}
	}
	}

and handles rebalancing correctly

[mingdyuo]

I modified it not to block processing like the example code, but same error still occurs time sometimes.

[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.

[meetgray]

ping?