Kafka Consumer every start reset offset to the specified location, can't continue to consume fr-CodePudding

Kafka in A topic (topic - A) start consumers are reset every time the offset to A fixed location, and the problem of topic - do not save A consumer group information, use the following command

./kafka - consumer - groups. Sh - the bootstrap - server localhost: 9092 - list

./kafka - consumer - groups. Sh - the bootstrap - server localhost: 9092 - go - group group - 5

When I open program to check the group information for:

 Consumer group 'group - 5' has no active members.

Closed view consumer groups list and view specify group do not check the information, return

 Error: Consumer group 'group - 5' does not exist.

Below is the Topic when start the program - A output information:

 O.A.K.C.C.I.C onsumerCoordinator - [Consumer clientId=Consumer - 1, groupId=group - 5] Revoking previously assigned partitions [] 
O.A.K.C.C.I.A bstractCoordinator - [Consumer clientId=Consumer - 1, groupId=group - 5] (Re -) joining the group 
O.A.K.C.C.I.A bstractCoordinator - [Consumer clientId=Consumer - 1, groupId=group - 5] Successfully joined the group with generation 1 
O.A.K.C.C.I.C onsumerCoordinator - [Consumer clientId=Consumer - 1, groupId=group - 5] Setting newly assigned partitions [A1] 0 
O.A.K.C.C onsumer. Internals. Fetcher - [Consumer clientId=Consumer - 1, groupId=group - 5] Resetting offset for partition A1-0 to offset 25143.

The last line of information:
Reference

O.A.K.C.C onsumer. The internals. The Fetcher - [Consumer clientId=Consumer - 1, groupId=group - 5] Resetting offset for partition A1-0 to offset 25143.

Each Topic - launched A will appear the line information, reset the offset to 25143,
If I run for a period of time, consumption to 25200, and then close the program, start again automatically reset to 25143 for me next time,
Multiple start is done with the group - 5.
With groups after start the program. The list command can see group - 5, use the describe command to check group - 5 shows Consumer group 'group - 5' has no active members.
After closing procedure in groups. The list can't see the group - 5, and use the describe returns the Error: Consumer group 'group - 5' does not exist.

Here is another Topic to start the normal output of information, not above the Resetting offset for partition information to offset XXX XXX, each can be launched from the last stop position to continue spending,

 org. Apache. Kafka. Clients. Metadata - Cluster ID: iQU30Fo1TViA2rkH9cxVYQ 
O.A.K.C.C.I.A bstractCoordinator - [Consumer clientId=Consumer - 3, groupId=wallet - 1] the Discovered group coordinator localhost: 9092 (id: 2147483646 rack: null) 
O.A.K.C.C.I.C onsumerCoordinator - [Consumer clientId=Consumer - 3, groupId=wallet - 1] Revoking previously assigned partitions [] 
O.A.K.C.C.I.A bstractCoordinator - [Consumer clientId=Consumer - 3, groupId=wallet - 1] (Re -) joining the group 
O.A.K.C.C.I.A bstractCoordinator - [Consumer clientId=Consumer - 3, groupId=wallet - 1] Successfully joined the group with generation 13 
O.A.K.C.C.I.C onsumerCoordinator - [Consumer clientId=Consumer - 3, groupId=wallet - 1] Setting newly assigned partitions [blockaddresscomplete3-0]

But can you tell me how to set the topic - A, let him to continue from the last stop position in each startup consumption, rather than every start reset to the fixed offset,

CodePudding user response:

refer to the original poster deantzzz response:

kafka in A topic (topic - A) start consumers are reset every time the offset to A fixed location, and the problem of topic - do not save A consumer group information, use the following command
./kafka - consumer - groups. Sh - the bootstrap - server localhost: 9092 - list 
./kafka - consumer - groups. Sh - the bootstrap - server localhost: 9092 - go - group group - 5 
When I open program to check the group information for:
 Consumer group 'group - 5' has no active members. 
Closed view consumer groups list and view specify group do not check the information, return
 Error: Consumer group 'group - 5' does not exist. 
Below is the Topic when start the program - A output information:
 O.A.K.C.C.I.C onsumerCoordinator - [Consumer clientId=Consumer - 1, groupId=group - 5] Revoking previously assigned partitions [] 
O.A.K.C.C.I.A bstractCoordinator - [Consumer clientId=Consumer - 1, groupId=group - 5] (Re -) joining the group 
O.A.K.C.C.I.A bstractCoordinator - [Consumer clientId=Consumer - 1, groupId=group - 5] Successfully joined the group with generation 1 
O.A.K.C.C.I.C onsumerCoordinator - [Consumer clientId=Consumer - 1, groupId=group - 5] Setting newly assigned partitions [A1] 0 
O.A.K.C.C onsumer. Internals. Fetcher - [Consumer clientId=Consumer - 1, groupId=group - 5] Resetting offset for partition A1-0 to offset 25143. 
The last line of information:
Reference
O.A.K.C.C onsumer. The internals. The Fetcher - [Consumer clientId=Consumer - 1, groupId=group - 5] Resetting offset for partition A1-0 to offset 25143.

Each Topic - launched A will appear the line information, reset the offset to 25143,
If I run for a period of time, consumption to 25200, and then close the program, start again automatically reset to 25143 for me next time,
Multiple start is done with the group - 5.
With groups after start the program. The list command can see group - 5, use the describe command to check group - 5 shows Consumer group 'group - 5' has no active members.
After closing procedure in groups. The list can't see the group - 5, and use the describe returns the Error: Consumer group 'group - 5' does not exist.

Here is another Topic to start the normal output of information, not above the Resetting offset for partition information to offset XXX XXX, each can be launched from the last stop position to continue spending,

org. Apache. Kafka. Clients. Metadata - Cluster ID: iQU30Fo1TViA2rkH9cxVYQ
O.A.K.C.C.I.A bstractCoordinator - [Consumer clientId=Consumer - 3, groupId=wallet - 1] the Discovered group coordinator localhost: 9092 (id: 2147483646 rack: null)
O.A.K.C.C.I.C onsumerCoordinator - [Consumer clientId=Consumer - 3, groupId=wallet - 1] Revoking previously assigned partitions []
nullnullnullnullnull