fix bug for cooperative-sticky consumer
Thanks tests!
The continue was wrong, so consumers that only invalidated some actually
kept nothing and never consumed again.
@@ -290,11 +290,30 @@ const (
assignSetMatching
)
+func (h assignHow) String() string {
+ switch h {
+ case assignWithoutInvalidating:
+ return "assign without invalidating"
+ case assignInvalidateAll:
+ return "assign invalidate all"
+ case assignInvalidateMatching:
+ return "assign invalidate matching"
+ case assignSetMatching:
+ return "assign set matching"
+ }
+ return ""
+}
+
// assignPartitions, called under the consumer's mu, is used to set new
// cursors or add to the existing cursors.
func (c *consumer) assignPartitions(assignments map[string]map[int32]Offset, how assignHow) {
seq := c.seq
+ c.cl.cfg.logger.Log(LogLevelInfo, "assigning partitions",
+ "assignments", assignments,
+ "how", how.String(),
+ )
+
if how != assignWithoutInvalidating {
// In this block, we immediately want to ensure that nothing
// currently buffered will be returned and that no active
@@ -318,10 +337,7 @@ func (c *consumer) assignPartitions(assignments map[string]map[int32]Offset, how
needsReset = false
} else {
if matchTopic, ok := assignments[usedPartition.cursor.topic]; ok {
- matchPartition, ok := matchTopic[usedPartition.cursor.partition]
- if !ok {
- continue
- }
+ if matchPartition, ok := matchTopic[usedPartition.cursor.partition]; ok {
needsReset = false
if how == assignInvalidateMatching {
usedPartition.cursor.setOffset(usedPartition.leaderEpoch, true, -1, -1, seq) // case 1
@@ -331,6 +347,7 @@ func (c *consumer) assignPartitions(assignments map[string]map[int32]Offset, how
}
}
}
+ }
if needsReset {
usedPartition.cursor.resetOffset(seq) // case 2
keep = append(keep, usedPartition)
@@ -347,6 +364,8 @@ func (c *consumer) assignPartitions(assignments map[string]map[int32]Offset, how
c.sourcesReadyMu.Unlock()
c.usingPartitions = keep
+
+ c.cl.cfg.logger.Log(LogLevelInfo, "assign maybe invalidated some partitions", "kept", keep)
}
// This assignment could contain nothing (for the purposes of
@@ -355,6 +374,8 @@ func (c *consumer) assignPartitions(assignments map[string]map[int32]Offset, how
return
}
+ c.cl.cfg.logger.Log(LogLevelInfo, "assign requires loading offsets")
+
// If we have a topic and partition loaded and the assignments use
// exact offsets, we can avoid looking up offsets.
waiting := offsetsWaitingLoad{
@@ -407,6 +428,8 @@ func (c *consumer) assignPartitions(assignments map[string]map[int32]Offset, how
}
}
+ c.cl.cfg.logger.Log(LogLevelInfo, "assign setting offsets to load", "to_load", assignments)
+
waiting.waitingList = assignments
if !waiting.isEmpty() {
waiting.mergeIntoLocked(c)
