I am seeing a strange issue with consumers that seem to refuse to consume data for particular partitions some of the time.
The symptom that I have observed is that an otherwise healthy consumer will just not consume any data from a given partition that was assigned to it (records from the affected partition are never returned when calling poll()). This does not occur on every rebalance, but when it does, it doesn't seem to recover until a subsequent rebalance. A partition that was perfectly healthy on the same consumer on the previous group generation can become unhealthy on the next generation.
The issue in #3106 seems potentially similar, but this has been occurring in 3.0.7 and 3.0.8. It may have been occurring earlier, but I didn't start testing 3.0.x before about 3.0.6. I can definitively say it didn't happen with 2.0.2, and I'm pretty sure it didn't happen with 2.2/2.3, but I have significantly better visibility into our releases using older versions that we're actively trying to upgrade (newer ones were only used in smaller deployments where introspection is harder).
Unfortunately, I am unable to paste raw/complete logs due to work restrictions, but I can relay some of what I have observed from enabling debug logging so far... I'm sorry - I know it makes it significantly harder to troubleshoot.
In one occurrence where only 2 partitions were assigned to a given consumer, one partition was drained as expected, and the other was ignored / impacted by this issue.
kafka.coordinator.consumer logged that it was setting newly assigned partitions for the group, indicating the 2 partitionskafka.consumer.fetcher periodically indicates that it is sending a ListOffsetRequest for the two partitions, and then that it is handling the ListOffsetResponse for each of them, but then it only sends fetch requests for the other (healthy) partition.- No logs are written about the partition being ignored until the next
ListOffsetRequest.
The logger levels that I have explicitly set:
kafka.cluster: INFOkafka.consumer.fetcher: DEBUGkafka.consumer.group: DEBUGkafka.coordinator: DEBUGkafka.coordinator.heartbeat: WARNINGkafka.metrics.metrics: INFOkafka.net: INFOkafka.protocol.parser: INFO
I know this isn't much to go on, but if there are other loggers that you think would provide more useful information, I can enable them.
I haven't been able to identify any particular setting or variable that seems to be a trigger to be able to reliably reproduce this in a given consumer, but I have been able to frequently reproduce it in a dev consumer group with 12 members. If I kill all of the containers and spawn new ones, by the time all of the instances are stable, there's usually 1 or 2 members that are impacted by this.
I am comfortable applying temporary patches to test workarounds/fixes or to get more information, if you think it may help. There are a couple areas in kafka.consumer.fetcher.Fetcher that seem like they may be potentially relevant, but I'm not familiar enough with this part of the code to have a good idea of where specifically something could be going wrong.
I am seeing a strange issue with consumers that seem to refuse to consume data for particular partitions some of the time.
The symptom that I have observed is that an otherwise healthy consumer will just not consume any data from a given partition that was assigned to it (records from the affected partition are never returned when calling
poll()). This does not occur on every rebalance, but when it does, it doesn't seem to recover until a subsequent rebalance. A partition that was perfectly healthy on the same consumer on the previous group generation can become unhealthy on the next generation.The issue in #3106 seems potentially similar, but this has been occurring in 3.0.7 and 3.0.8. It may have been occurring earlier, but I didn't start testing 3.0.x before about 3.0.6. I can definitively say it didn't happen with 2.0.2, and I'm pretty sure it didn't happen with 2.2/2.3, but I have significantly better visibility into our releases using older versions that we're actively trying to upgrade (newer ones were only used in smaller deployments where introspection is harder).
Unfortunately, I am unable to paste raw/complete logs due to work restrictions, but I can relay some of what I have observed from enabling debug logging so far... I'm sorry - I know it makes it significantly harder to troubleshoot.
In one occurrence where only 2 partitions were assigned to a given consumer, one partition was drained as expected, and the other was ignored / impacted by this issue.
kafka.coordinator.consumerlogged that it was setting newly assigned partitions for the group, indicating the 2 partitionskafka.consumer.fetcherperiodically indicates that it is sending aListOffsetRequestfor the two partitions, and then that it is handling theListOffsetResponsefor each of them, but then it only sends fetch requests for the other (healthy) partition.ListOffsetRequest.The logger levels that I have explicitly set:
kafka.cluster: INFOkafka.consumer.fetcher: DEBUGkafka.consumer.group: DEBUGkafka.coordinator: DEBUGkafka.coordinator.heartbeat: WARNINGkafka.metrics.metrics: INFOkafka.net: INFOkafka.protocol.parser: INFOI know this isn't much to go on, but if there are other loggers that you think would provide more useful information, I can enable them.
I haven't been able to identify any particular setting or variable that seems to be a trigger to be able to reliably reproduce this in a given consumer, but I have been able to frequently reproduce it in a dev consumer group with 12 members. If I kill all of the containers and spawn new ones, by the time all of the instances are stable, there's usually 1 or 2 members that are impacted by this.
I am comfortable applying temporary patches to test workarounds/fixes or to get more information, if you think it may help. There are a couple areas in
kafka.consumer.fetcher.Fetcherthat seem like they may be potentially relevant, but I'm not familiar enough with this part of the code to have a good idea of where specifically something could be going wrong.