У меня есть приложение KafkaStream, которое раньше отлично работало. Теперь независимо от того, сколько раз я перезапускаю его с новым идентификатором приложения, он никогда не начинает потреблять topi c, и я получаю этот журнал:
INFO org.apache.kafka.streams.KafkaStreams stream-client [score_redeemX-7e37d43e-12ab-4c66-984e-5f959d4e5e08] State transition from REBALANCING to RUNNING
INFO org.apache.kafka.clients.consumer.KafkaConsumer [Consumer clientId=score_redeemX-7e37d43e-12ab-4c66-984e-5f959d4e5e08-StreamThread-1-restore-consumer, groupId=null] Unsubscribed all topics or patterns and assigned partitions
INFO org.apache.kafka.streams.processor.internals.StreamThread stream-thread [score_redeemX-7e37d43e-12ab-4c66-984e-5f959d4e5e08-StreamThread-1] State transition from PARTITIONS_ASSIGNED to RUNNING
INFO org.apache.kafka.clients.consumer.KafkaConsumer [Consumer clientId=score_redeemX-7e37d43e-12ab-4c66-984e-5f959d4e5e08-StreamThread-1-restore-consumer, groupId=null] Unsubscribed all topics or patterns and assigned partitions
Когда я устанавливаю уровень журнала для отладки, это журнал приложения производит:
2020-05-09 18:45:20,335 DEBUG org.apache.kafka.streams.processor.internals.StreamThread stream-thread [score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-2] Committed all active tasks [] and standby tasks [] in 0ms
2020-05-09 18:45:20,338 DEBUG org.apache.kafka.streams.processor.internals.StreamThread stream-thread [score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-1] Committed all active tasks [] and standby tasks [] in 0ms
2020-05-09 18:45:20,339 DEBUG org.apache.kafka.streams.processor.internals.StreamThread stream-thread [score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-3] Committed all active tasks [] and standby tasks [] in 0ms
2020-05-09 18:45:21,339 DEBUG org.apache.kafka.streams.processor.internals.StreamThread stream-thread [score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-2] Committed all active tasks [] and standby tasks [] in 0ms
2020-05-09 18:45:21,341 DEBUG org.apache.kafka.streams.processor.internals.StreamThread stream-thread [score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-1] Committed all active tasks [] and standby tasks [] in 0ms
2020-05-09 18:45:21,342 DEBUG org.apache.kafka.streams.processor.internals.StreamThread stream-thread [score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-3] Committed all active tasks [] and standby tasks [] in 0ms
2020-05-09 18:45:22,343 DEBUG org.apache.kafka.streams.processor.internals.StreamThread stream-thread [score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-2] Committed all active tasks [] and standby tasks [] in 0ms
2020-05-09 18:45:22,344 DEBUG org.apache.kafka.streams.processor.internals.StreamThread stream-thread [score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-1] Committed all active tasks [] and standby tasks [] in 0ms
2020-05-09 18:45:22,345 DEBUG org.apache.kafka.streams.processor.internals.StreamThread stream-thread [score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-3] Committed all active tasks [] and standby tasks [] in 0ms
2020-05-09 18:45:22,435 DEBUG org.apache.kafka.clients.consumer.internals.AbstractCoordinator [Consumer clientId=score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-3-consumer, groupId=score_redeem] Sending Heartbeat request to coordinator ###.###.###.###:9092 (id: 2147483642 rack: null)
2020-05-09 18:45:22,437 DEBUG org.apache.kafka.clients.consumer.internals.AbstractCoordinator [Consumer clientId=score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-3-consumer, groupId=score_redeem] Received successful Heartbeat response
2020-05-09 18:45:22,465 DEBUG org.apache.kafka.clients.consumer.internals.AbstractCoordinator [Consumer clientId=score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-2-consumer, groupId=score_redeem] Sending Heartbeat request to coordinator ###.###.###.###:9092 (id: 2147483642 rack: null)
2020-05-09 18:45:22,466 DEBUG org.apache.kafka.clients.consumer.internals.AbstractCoordinator [Consumer clientId=score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-2-consumer, groupId=score_redeem] Received successful Heartbeat response
2020-05-09 18:45:22,468 DEBUG org.apache.kafka.clients.consumer.internals.AbstractCoordinator [Consumer clientId=score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-1-consumer, groupId=score_redeem] Sending Heartbeat request to coordinator ###.###.###.###:9092 (id: 2147483642 rack: null)
2020-05-09 18:45:22,470 DEBUG org.apache.kafka.clients.consumer.internals.AbstractCoordinator [Consumer clientId=score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-1-consumer, groupId=score_redeem] Received successful Heartbeat response
Никакая задача не назначена, как показывает журнал:
2020-05-09 18:43:04,322 DEBUG org.apache.kafka.streams.processor.internals.StreamsPartitionAssignor stream-thread [score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-2-consumer] Constructed client metadata {0633c837-8b94-4ba3-8e59-67a4c71de337=ClientMetadata{hostInfo=null, consumers=[score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-2-consumer-b830a8e5-8654-40b7-8142-d1754c32268a], state=[activeTasks: ([]) standbyTasks: ([]) assignedTasks: ([]) prevActiveTasks: ([]) prevStandbyTasks: ([]) prevAssignedTasks: ([]) capacity: 1]}} from the member subscriptions.
Это часть журнала, в которой отображается информация о лидере и члене:
2020-05-09 18:43:04,361 DEBUG org.apache.kafka.clients.consumer.internals.AbstractCoordinator [Consumer clientId=score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-3-consumer, groupId=score_redeem] Received successful JoinGroup response: JoinGroupResponse(throttleTimeMs=0, error=NONE, generationId=4, groupProtocol=stream, memberId=score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-3-consumer-93905915-8b52-44b4-afd1-2a5c79a272a4, leaderId=score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-2-consumer-b830a8e5-8654-40b7-8142-d1754c32268a, members=)
2020-05-09 18:43:04,362 DEBUG org.apache.kafka.clients.consumer.internals.AbstractCoordinator [Consumer clientId=score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-3-consumer, groupId=score_redeem] Sending follower SyncGroup to coordinator ###.###.###.###:9092 (id: 2147483642 rack: null): (type=SyncGroupRequest, groupId=score_redeem, generationId=4, memberId=score_redeem-0633c837-8b94-4ba3-8e59-67a4c71de337-StreamThread-3-consumer-93905915-8b52-44b4-afd1-2a5c79a272a4, groupAssignment=)
У меня работает много приложений, и все они испытывают это. Я пробовал делать новые разъемы, но все равно не повезло. Но я могу успешно использовать эти темы с помощью kafka-console-consumer.
Еще один странный, но очень важный момент - когда я переписывал приложение с помощью Kafka Consumer API, оно работало нормально !!!