Kafka reader gets error: read tcp xxx.xxx.xxx.xxx:xxxx->xxx.xxx.xxx.xxx:xxxx: i/o timeout

[BlasterAlex]

Describe the bug

Hi, I get an error when reading from the topic. The error occurs with some periodicity, not all the time. After it occurs, kafka reader recreates the connection and continues to work.

Kafka Version

kafka version: v.2.8.1
kafka-go version: v0.4.35

To Reproduce

Here are my sample kafka reader settings:

mechanism, _ := scram.Mechanism(scram.SHA512, username, password)

dialer := &kafka.Dialer{
	Timeout:       10 * time.Second,
	DualStack:     true,
	SASLMechanism: mechanism,
}

topic := kafka.TopicConfig{
	Topic:             topicName,
	NumPartitions:     3,
	ReplicationFactor: 3,
	ConfigEntries: []kafka.ConfigEntry{
		{
			ConfigName:  "retention.bytes",
			ConfigValue: strconv.Itoa(-1),
		},
		{
			ConfigName:  "retention.ms",
			ConfigValue: strconv.Itoa(int((24 * time.Hour).Milliseconds())),
		},
	},
}

nr := kafka.NewReader(kafka.ReaderConfig{
	Brokers:               []string{kafkaAddress},
	Topic:                 topic.Topic,
	GroupID:               kafkaGroupName,
	WatchPartitionChanges: true,
	Dialer:                dialer,
	MaxWait:               10 * time.Second,
	MinBytes:              10e3, // 10KiB
	MaxBytes:              10e6, // 10MiB
	CommitInterval:        time.Second,
	Logger:                &debugLogger{},
	ErrorLogger:           &errorLogger{},
})

Expected Behavior

No "i/o timeout" error and reconnecting to the kafka topic.

Observed Behavior

The kafka reader logs when a connection timeout error occurs:

[2022-09-22T10:37:24.113][level=ERROR]: the kafka reader got an unknown error reading partition 2 of gt_events at offset 29055: read tcp xxx.xxx.xxx.xxx:xxxx->xxx.xxx.xxx.xxx:xxxx: i/o timeout
[2022-09-22T10:37:24.12][level=ERROR]: the kafka reader got an unknown error reading partition 1 of ht_events at offset 158: read tcp xxx.xxx.xxx.xxx:xxxx->xxx.xxx.xxx.xxx:xxxx: i/o timeout
[2022-09-22T10:37:24.213][level=DEBUG]: initializing kafka reader for partition 2 of gt_events starting at offset 29055
[2022-09-22T10:37:24.221][level=DEBUG]: initializing kafka reader for partition 1 of ht_events starting at offset 158
[2022-09-22T10:37:24.506][level=DEBUG]: the kafka reader for partition 1 of ht_events is seeking to offset 158
[2022-09-22T10:37:24.506][level=DEBUG]: the kafka reader for partition 2 of gt_events is seeking to offset 29055
[2022-09-22T10:44:14.618][level=ERROR]: the kafka reader got an unknown error reading partition 0 of rt_events at offset 35251: read tcp xxx.xxx.xxx.xxx:xxxx->xxx.xxx.xxx.xxx:xxxx: i/o timeout
[2022-09-22T10:44:14.718][level=DEBUG]: initializing kafka reader for partition 0 of rt_events starting at offset 35251
[2022-09-22T10:44:14.812][level=DEBUG]: the kafka reader for partition 0 of rt_events is seeking to offset 35251

In normal case, I have this message by ReaderConfig.MaxWait timeout:

[2022-09-23T14:21:03.835][level=DEBUG]: no messages received from kafka within the allocated time for partition 0 of rt_events at offset 162

Additional Context

I was able to get the type of error from the log message. It is *net.OpError, which contains *poll.DeadlineExceededError.

I could not find in the code when the connection timeout triggered, as well as the config attribute that is responsible for this.

Deadline for the batch reader is set here: https://github.com/segmentio/kafka-go/blob/v0.4.35/reader.go#L1478

This sets the deadline for reading the message, which is equal to MaxWait - time.Second, as far as I understood from the code: https://github.com/segmentio/kafka-go/blob/v0.4.35/conn.go#L791

How can I increase the timeout of kafka Conn or decrease the timeout of reading the message? I think the problem is that the tcp connection is closed before the reading from the topic is done.

I will be glad to get any advice and suggestions :)

[rhansen2]

HI @BlasterAlex ,

Do you still see these periodic timeouts if you increase your MaxWait configuration?

[BlasterAlex]

Is there any way I can control the timeout of the net.Conn that is used within kafka.Conn for the tcp connection?

As far as I understand, it is the deadline inside net.Conn that triggers in my case, because the connection timeout kafka (RequestTimedOut) is caught in the read function.

I have tried increasing the values of kafka.Dialer Timeout and KeepAlive, but this has no effect on the error.

[BlasterAlex]

Hi, @rhansen2, the error was reproduced with the value of MaxWait = 1m.

I also tried with values of 2m and 5m. Now I periodically get the following error:

[2022-09-27T14:46:27.21][level=DEBUG]: stopped heartbeat for group group-name
[2022-09-27T14:46:27.21][level=ERROR]: Problem getting partitions while checking for changes, read tcp xxx.xxx.xxx.xxx:xxxx->xxx.xxx.xxx.xxx:xxxx: use of closed network connection
[2022-09-27T14:46:27.21][level=DEBUG]: stopped commit for group group-name
[2022-09-27T14:46:27.21][level=DEBUG]: stopped partition watcher for group, group-name, topic qa-3_events
[2022-09-27T14:46:27.211][level=DEBUG]: stopped heartbeat for group group-name
[2022-09-27T14:46:27.211][level=DEBUG]: stopped heartbeat for group group-name
[2022-09-27T14:46:27.211][level=ERROR]: Problem getting partitions while checking for changes, read tcp xxx.xxx.xxx.xxx:xxxx->xxx.xxx.xxx.xxx:xxxx: use of closed network connection
[2022-09-27T14:46:27.211][level=DEBUG]: stopped partition watcher for group, group-name, topic qa-4_events
[2022-09-27T14:46:27.211][level=DEBUG]: stopped commit for group group-name
[2022-09-27T14:46:27.211][level=DEBUG]: stopped commit for group group-name
[2022-09-27T14:46:27.211][level=ERROR]: Problem getting partitions while checking for changes, read tcp xxx.xxx.xxx.xxx:xxxx->xxx.xxx.xxx.xxx:xxxx: use of closed network connection
[2022-09-27T14:46:27.211][level=DEBUG]: stopped partition watcher for group, group-name, topic qa-5_events

and

[2022-09-27T15:28:25.75][level=ERROR]: Problem getting partitions while checking for changes, read tcp xxx.xxx.xxx.xxx:xxxx->xxx.xxx.xxx.xxx:xxxx: i/o timeout
[2022-09-27T15:28:25.75][level=DEBUG]: stopped partition watcher for group, group-name, topic qa-3_events
[2022-09-27T15:28:25.75][level=DEBUG]: stopped heartbeat for group group-name
[2022-09-27T15:28:25.75][level=DEBUG]: stopped commit for group group-name
[2022-09-27T15:28:25.812][level=ERROR]: Problem getting partitions while checking for changes, read tcp xxx.xxx.xxx.xxx:xxxx->xxx.xxx.xxx.xxx:xxxx: i/o timeout
[2022-09-27T15:28:25.812][level=DEBUG]: stopped partition watcher for group, group-name, topic qa-4_events
[2022-09-27T15:28:25.812][level=ERROR]: Problem getting partitions while checking for changes, read tcp xxx.xxx.xxx.xxx:xxxx->xxx.xxx.xxx.xxx:xxxx: i/o timeout
[2022-09-27T15:28:25.812][level=DEBUG]: stopped commit for group group-name
[2022-09-27T15:28:25.812][level=DEBUG]: stopped partition watcher for group, group-name, topic qa-5_events
[2022-09-27T15:28:25.812][level=DEBUG]: stopped heartbeat for group group-name
[2022-09-27T15:28:25.813][level=DEBUG]: stopped commit for group group-name
[2022-09-27T15:28:25.813][level=DEBUG]: stopped heartbeat for group group-name

after that the consumer group was recreated

[BlasterAlex]

I also noticed a peculiarity that the error occurs precisely when the kafka reader is not in the gap 1 sec:

[2022-09-27T15:40:03.714][level=DEBUG][class=kfk:logger.go:10]: no messages received from kafka within the allocated time for partition 1 of qa-3_events at offset 27014
[2022-09-27T15:40:03.808][level=DEBUG][class=kfk:logger.go:10]: no messages received from kafka within the allocated time for partition 1 of qa-5_events at offset 86
[2022-09-27T15:40:03.823][level=DEBUG][class=kfk:logger.go:10]: no messages received from kafka within the allocated time for partition 2 of qa-3_events at offset 44277
[2022-09-27T15:40:03.823][level=DEBUG][class=kfk:logger.go:10]: no messages received from kafka within the allocated time for partition 0 of qa-4_events at offset 8377
[2022-09-27T15:40:03.919][level=DEBUG][class=kfk:logger.go:10]: no messages received from kafka within the allocated time for partition 2 of qa-5_events at offset 54
[2022-09-27T15:40:03.919][level=DEBUG][class=kfk:logger.go:10]: no messages received from kafka within the allocated time for partition 2 of qa-4_events at offset 8304
[2022-09-27T15:40:04.811][level=ERROR][class=kfk:logger.go:16]: the kafka reader got an unknown error reading partition 0 of qa-5_events at offset 93: read tcp xxx.xxx.xxx.xxx:xxxx->xxx.xxx.xxx.xxx:xxxx: i/o timeout
[2022-09-27T15:40:04.812][level=ERROR][class=kfk:logger.go:16]: the kafka reader got an unknown error reading partition 1 of qa-4_events at offset 7896: read tcp xxx.xxx.xxx.xxx:xxxx->xxx.xxx.xxx.xxx:xxxx: i/o timeout
[2022-09-27T15:40:04.813][level=ERROR][class=kfk:logger.go:16]: the kafka reader got an unknown error reading partition 0 of qa-3_events at offset 15219: read tcp xxx.xxx.xxx.xxx:xxxx->xxx.xxx.xxx.xxx:xxxx: i/o timeout
[2022-09-27T15:40:04.912][level=DEBUG][class=kfk:logger.go:10]: initializing kafka reader for partition 1 of qa-4_events starting at offset 7896
[2022-09-27T15:40:04.912][level=DEBUG][class=kfk:logger.go:10]: initializing kafka reader for partition 0 of qa-5_events starting at offset 93
[2022-09-27T15:40:04.914][level=DEBUG][class=kfk:logger.go:10]: initializing kafka reader for partition 0 of qa-3_events starting at offset 15219

As you can see from the logs, the readers that hit the MaxWait timeout successfully handled the RequestTimedOut error, but those that were delayed by 1 sec (defaultRTT = 1 * time.Second) fell down with a tcp error.

[rhansen2]

Based on your log output it looks like you're getting timeouts in various locations; Both partition watching and reading. For the timeouts reading messages I wonder if we need to pass MaxWait here so that it respects your configured max wait instead of using the any default timeouts.

[rhansen2]

Potentially related to #989

[rhansen2]

With #989 merged is that potentially a fix for your issue?

[BlasterAlex]

Hi, I hope so. I stopped looking for the cause of the problem for now, maybe it's a problem of slow kafka on my environment

[rhansen2]

In that case I'm going to close for now, feel free to reopen or create a new issue in the future, thanks!