{"id":9723,"date":"2023-01-06T15:50:54","date_gmt":"2023-01-06T20:50:54","guid":{"rendered":"https:\/\/www.rushworth.us\/lisa\/?p=9723"},"modified":"2023-01-10T14:17:54","modified_gmt":"2023-01-10T19:17:54","slug":"kafka-reset-__consumer_offsets-topic","status":"publish","type":"post","link":"https:\/\/www.rushworth.us\/lisa\/?p=9723","title":{"rendered":"Kafka: Reset __consumer_offsets Topic"},"content":{"rendered":"

I set up a few topics for a new process, but the client was continually unable to subscribe getting the following messages:<\/p>\n

[2023-01-06 12:54:58,965] INFO [Worker clientId=connect-1, groupId=connect-cluster-group-dev] Group coordinator kafkahost.example.com:9092 (id: 2147483415 rack: null) is unavailable or invalid due to cause: error response NOT_COORDINATOR.isDisconnected: false. Rediscovery will be attempted. (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:904)\r\n[2023-01-06 12:54:58,966] INFO [Worker clientId=connect-1, groupId=connect-cluster-group-dev] JoinGroup failed: This is not the correct coordinator. Marking coordinator unknown. Sent generation was Generation{generationId=-1, memberId='', protocol='null'} (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:623)\r\n[2023-01-06 12:54:58,966] INFO [Worker clientId=connect-1, groupId=connect-cluster-group-dev] Rebalance failed. (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:472)\r\norg.apache.kafka.common.errors.NotCoordinatorException: This is not the correct coordinator.<\/pre>\n
The output should<\/em> contain your topic and an offset:<\/p>\n
kafkasandbox:bin # .\/kafka-console-consumer.sh  --bootstrap-server kafkaserver.example.com:9092 --formatter \"kafka.coordinator.group.GroupMetadataManager\\$OffsetsMessageFormatter\" --topic __consumer_offsets\r\n[APP_RUN,appa_DB_error,5]::OffsetAndMetadata(offset=912219, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1673378016876, expireTimestamp=None)\r\n[APP_RUN,appb_DB_error,5]::OffsetAndMetadata(offset=424539, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1673378016938, expireTimestamp=None)\r\n[APP_RUN,appb_DB_error,3]::OffsetAndMetadata(offset=359340, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1673378016938, expireTimestamp=None)\r\n[APP_RUN,appc_DB_error,2]::OffsetAndMetadata(offset=986361, leaderEpoch=Optional.empty, metadata=, commitTimestamp=1673378017005, expireTimestamp=None)\r\n<\/pre>\n
I recreated the topics with the replication factor equal to the number of servers. I restarted all of the Kafka and ZooKeeper services. We tried from other clients. We tried searching the internet for some pointer as to what has gone awry. The only thing I could find is that the __consumer_offsets topic wasn’t right … you should be able to read the offset data:<\/p>\n
\/kafka-console-consumer.sh --bootstrap-server kafkahost.example.net:9092 --formatter \"kafka.coordinator.group.GroupMetadataManager\\$OffsetsMessageFormatter\" --topic __consumer_offsets<\/span><\/pre>\n
Now, resetting the entire consumer offsets partition is a massively<\/em> bad idea in a production environment. We’ve had to advance offsets before, but never had to reset the entire partition.<\/p>\n
The documentation I found online used a ZooKeeper command that no longer exists — ‘rmr’ — but deleteall (basically a subtree deletion) did the trick.<\/p>\n
# Reset consumer offsets\r\n# Note -- this is a *really* bad idea in production as it can lead to missed messages. You should record the current offsets, clear the topic, then set the offset for the client groups\r\nsystemctl stop kafka # on all servers\r\nmv \/kafka\/ws_npm_kafka\/data-kafka\/__consumer_offsets-* \/tmp\/\r\n.\/zookeeper-shell.sh localhost:2181\r\ndeleteall \/brokers\/topics\/__consumer_offsets\r\n# Once this has been done on all servers, restart kafka\r\nsystemctl start kafka<\/pre>\n","protected":false},"excerpt":{"rendered":"
I set up a few topics for a new process, but the client was continually unable to subscribe getting the following messages: [2023-01-06 12:54:58,965] INFO [Worker clientId=connect-1, groupId=connect-cluster-group-dev] Group coordinator kafkahost.example.com:9092 (id: 2147483415 rack: null) is unavailable or invalid due to cause: error response NOT_COORDINATOR.isDisconnected: false. Rediscovery will be attempted. (org.apache.kafka.clients.consumer.internals.AbstractCoordinator:904) [2023-01-06 12:54:58,966] INFO [Worker …<\/p>\n","protected":false},"author":1,"featured_media":0,"comment_status":"open","ping_status":"open","sticky":false,"template":"","format":"standard","meta":{"footnotes":""},"categories":[1762],"tags":[1361],"class_list":["post-9723","post","type-post","status-publish","format-standard","hentry","category-kafka","tag-kafka"],"_links":{"self":[{"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/posts\/9723","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/users\/1"}],"replies":[{"embeddable":true,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=%2Fwp%2Fv2%2Fcomments&post=9723"}],"version-history":[{"count":3,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/posts\/9723\/revisions"}],"predecessor-version":[{"id":9729,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=\/wp\/v2\/posts\/9723\/revisions\/9729"}],"wp:attachment":[{"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=%2Fwp%2Fv2%2Fmedia&parent=9723"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=%2Fwp%2Fv2%2Fcategories&post=9723"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/www.rushworth.us\/lisa\/index.php?rest_route=%2Fwp%2Fv2%2Ftags&post=9723"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}