fix: comments and logs

Signed-off-by: napallday <bzx0619@gmail.com>

consumer_group.go

@@ -378,11 +378,12 @@ func (c *consumerGroup) start(ctx context.Context, topics []string, handlerV2 Co
 	newAssignedPartitions := diffAssignment(claims, c.ownedPartitions)
 	if c.protocol == COOPERATIVE {
 		revokedPartitions := diffAssignment(c.ownedPartitions, claims)
-		Logger.Printf("updating consumer(group:%s, member:%s, generation:%d)\n"+
+		Logger.Printf("updating consumer(group:%s, member:%s, generation:%d, isLeader:%v)\n"+
 			"All Assignments: %v\n"+
 			"New Partitions: %v\n"+
 			"Revoked Partitions: %v\n",
-			c.groupID, c.memberID, c.generationID, claims, newAssignedPartitions, revokedPartitions)
+			c.groupID, c.memberID, c.generationID, c.isLeader,
+			claims, newAssignedPartitions, revokedPartitions)
 
 		if len(revokedPartitions) > 0 {
 			err = c.revokedPartitions(revokedPartitions)
@@ -410,7 +411,8 @@ func (c *consumerGroup) start(ctx context.Context, topics []string, handlerV2 Co
 		cancelFunc()
 	case <-ctx.Done():
 	}
-	Logger.Printf("consumer(group:%s, member:%s, generation:%d) context is done", c.groupID, c.memberID, c.generationID)
+	Logger.Printf("consumer(group:%s, member:%s, generation:%d, isLeader:%v) context is done\n",
+		c.groupID, c.memberID, c.generationID, c.isLeader)
 
 	// if using EAGER rebalance protocol, we need to revoke all owned partitions before sending new JoinGroupRequest
 	if c.protocol == EAGER {
@@ -1095,7 +1097,8 @@ func (c *consumerGroup) topicToPartitionNumbers(topics []string) (map[string]int
 }
 
 func (c *consumerGroup) revokedPartitions(revokedPartitions map[string][]int32) error {
-	Logger.Printf("revoking partitions: %v", c.groupID, revokedPartitions)
+	Logger.Printf("consumer(group:%s, member:%s, generation:%d, isLeader:%v) revoking partitions: %v\n",
+		c.groupID, c.memberID, c.generationID, c.isLeader, revokedPartitions)
 
 	// close revoked partition consumers
 	c.removeClaims(revokedPartitions)
@@ -1121,13 +1124,15 @@ func (c *consumerGroup) revokedOwnedPartitions() {
 }
 
 func (c *consumerGroup) startNewPartitions(newAssignedPartitions map[string][]int32) error {
+	Logger.Printf("consumer(group:%s, member:%s, generation:%d, isLeader:%v) starting new assigned partitions: %v",
+		c.groupID, c.memberID, c.generationID, c.isLeader, newAssignedPartitions)
+
 	// create partition offset managers for each new assigned partitions
 	for topic, partitions := range newAssignedPartitions {
 		for _, partition := range partitions {
 			pom, err := c.offsetManager.ManagePartition(topic, partition)
 			if err != nil {
 				Logger.Printf("unable to create partition offset manager for %s/%d, err: %v", topic, partition, err)
-				// todo maybe close all offset managers here
 				return err
 			}
 
@@ -1562,6 +1567,7 @@ type ConsumerGroupHandlerV2 interface {
 	// Setup runs at the beginning of setting up new assigned partitions, before ConsumeClaim.
 	// For EAGER rebalance strategy, this is to set up all assigned partitions.
 	// For COOPERATIVE rebalance strategy, this is only to set up new assigned partitions.
+	// Note that even if there are no new assigned partitions, this method will still be called after rebalance.
 	Setup(offsetManger OffsetManager, newAssignedPartitions map[string][]int32)
 
 	// Cleanup runs after ConsumeClaim, but before the offsets are committed for the claim.

examples/consumergroup_cooperative/main.go

@@ -34,8 +34,6 @@ func init() {
 	flag.StringVar(&assignors, "assignors", "cooperative-sticky", "Consumer group partition assignment strategies (range, roundrobin, sticky, cooperative-sticky)")
 	flag.BoolVar(&oldest, "oldest", true, "Kafka consumer consume initial offset from oldest")
 	flag.BoolVar(&verbose, "verbose", false, "Sarama logging")
-	// flag string slice
-	// https://stackoverflow.com/questions/1752414/how-to-receive-variable-number-of-arguments-as-a-command-line-flag
 
 	flag.Parse()
 
@@ -170,15 +168,16 @@ type Consumer struct{}
 // Setup runs at the beginning of setting up new assigned partitions, before ConsumeClaim.
 // For EAGER rebalance strategy, this is to set up all assigned partitions.
 // For COOPERATIVE rebalance strategy, this is only to set up new assigned partitions.
+// Note that even if there are no new assigned partitions, this method will still be called after rebalance.
 func (consumer *Consumer) Setup(offsetManger sarama.OffsetManager, newAssignedPartitions map[string][]int32) {
-	log.Printf("newAssignedPartitions: %v", newAssignedPartitions)
+	log.Printf("[Setup] newAssignedPartitions: %v", newAssignedPartitions)
 }
 
 // Cleanup runs after ConsumeClaim, but before the offsets are committed for the claim.
 // For EAGER rebalance strategy, this is to clean up all assigned partitions.
 // For COOPERATIVE rebalance strategy, this is only to clean up revoked partitions.
 func (consumer *Consumer) Cleanup(offsetManger sarama.OffsetManager, revokedPartitions map[string][]int32) {
-	log.Printf("revokedPartitions: %v", revokedPartitions)
+	log.Printf("[Cleanup] revokedPartitions: %v", revokedPartitions)
 }
 
 // ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().