diff --git a/bulker/bulkerapp/app/abstract_batch_consumer.go b/bulker/bulkerapp/app/abstract_batch_consumer.go index 7ad98c2dd..4381aad77 100644 --- a/bulker/bulkerapp/app/abstract_batch_consumer.go +++ b/bulker/bulkerapp/app/abstract_batch_consumer.go @@ -26,6 +26,12 @@ const errorHeader = "error" const pauseHeartBeatInterval = 120 * time.Second +// assignmentFailuresBeforeRestart - consecutive runs a retry consumer may end +// without a partition assignment before it is treated as having lost its group +// membership. More than one, so an ordinary rebalance is not mistaken for a +// zombie (see ConsumeAll). +const assignmentFailuresBeforeRestart = 3 + type BatchSizesFunction func(*bulker.StreamOptions) (batchSize int, batchSizeBytes int, retryBatchSize int) type BatchFunction func(destination *Destination, batchNum, batchSize, batchSizeBytes, retryBatchSize int, highOffset int64, updatedHighOffset int) (counters BatchCounters, state bulker.State, nextBatch bool, err error) type ShouldConsumeFunction func(partitionId int32, committedOffset, highOffset int64) bool @@ -74,6 +80,30 @@ type AbstractBatchConsumer struct { //restarting guards against piling up overlapping restartConsumer calls //from the pause heartbeat loop (see restartConsumerAsync). restarting atomic.Bool + //restartMu serializes restartConsumer (see there). + restartMu sync.Mutex + //consumerCreatedAt is when the current kafka consumer was created (unix + //nanos). A consumer needs a poll to join the group and get its assignment, + //so membership checks only apply once it is older than a grace period. + consumerCreatedAt atomic.Int64 + //restartGeneration counts restarts, which suffix the static + //group.instance.id so a replacement is never fenced against the consumer + //it replaces (see newConsumer). + restartGeneration atomic.Int64 + //lastRestartAt is when a restart last published a replacement consumer + //(unix nanos; zero before the first one). Only a replacement is subject to + //the restart cooldown — see restartConsumer. + lastRestartAt atomic.Int64 + //assignmentFailures counts consecutive runs that ended without a partition + //assignment (retry mode). One is a rebalance; several in a row is a zombie. + assignmentFailures atomic.Int64 + //closedConsumers records every *kafka.Consumer this object has closed. + //Several paths may hold the same pointer (suspend, restart quarantine, + //close), and confluent-kafka-go panics on a second Close, so all closes go + //through closeConsumer. One entry per restart; restarts are rare, so the + //set is never pruned. + closedMu sync.Mutex + closedConsumers map[*kafka.Consumer]struct{} batchSizeFunc BatchSizesFunction batchFunc BatchFunction @@ -130,7 +160,8 @@ func NewAbstractBatchConsumer(repository *Repository, destinationId string, batc //restartConsumer); the heartbeat picks it up on the next pass. resumeChannel: make(chan struct{}, 1), //unbuffered: suspend must rendezvous with the heartbeat (see field doc). - stopChannel: make(chan struct{}), + stopChannel: make(chan struct{}), + closedConsumers: map[*kafka.Consumer]struct{}{}, } bc.idle.Store(true) return bc, nil @@ -275,7 +306,7 @@ func (bc *AbstractBatchConsumer) ConsumeAll() (counters BatchCounters, err error } maxBatchSize, maxBatchSizeBytes, retryBatchSize := bc.batchSizeFunc(streamOptions) - consumer, err := bc.initConsumer(false) + consumer, created, err := bc.initConsumer() if err != nil { bc.errorMetric("resume_error") return BatchCounters{}, bc.NewError("Failed to resume kafka consumer: %v", err) @@ -302,27 +333,59 @@ func (bc *AbstractBatchConsumer) ConsumeAll() (counters BatchCounters, err error } if err != nil || len(ass) != 1 { bc.errorMetric("assignment_error") + //A single failure is not treated as lost membership (JITSU-214): + //retry consumers of one topic share a group across the fleet, one + //partition each, so missing an assignment in this window is routine + //during a rebalance, and restarting would rejoin under a new + //instance id and force another fleet-wide rebalance. A rebalance + //settles within a session timeout, so several runs in a row without + //an assignment is a zombie rather than churn. + if !created && bc.assignmentFailures.Add(1) >= assignmentFailuresBeforeRestart { + bc.membershipLost(fmt.Sprintf("no partition assignment in %d consecutive runs", assignmentFailuresBeforeRestart)) + bc.assignmentFailures.Store(0) + bc.restartConsumer(nil) + } return BatchCounters{}, bc.NewError("Failed to get consumer assignment (%d): %v", len(ass), err) } + bc.assignmentFailures.Store(0) partition = ass[0].Partition bc.Infof("Assigned partition: %d", partition) } _, highOffset, err = consumer.QueryWatermarkOffsets(bc.topicId, partition, 10_000) updatedHighOffset = highOffset offsets, erro := consumer.Committed([]kafka.TopicPartition{{Topic: &bc.topicId, Partition: partition}}, 10_000) - if len(offsets) > 0 { - if offsets[0].Offset != kafka.OffsetInvalid { - commitedOffset = int64(offsets[0].Offset) - } else { - bc.Errorf("Failed to query commited offsets.") - } + if erro != nil { + bc.errorMetric("query_committed_failed") + bc.Errorf("Failed to query committed offsets: %v", erro) + } else if len(offsets) > 0 && offsets[0].Offset != kafka.OffsetInvalid { + commitedOffset = int64(offsets[0].Offset) } else { - bc.Errorf("Failed to query commited offsets: %v", erro) + //Not an error by itself: a brand-new topic has no committed offset until + //its first batch. It is also what a consumer whose group was deleted + //sees — that case is caught by the membership check below. + bc.Infof("No committed offset for the consumer group yet. High watermark: %d", highOffset) } if err != nil { bc.errorMetric("query_watermark_failed") return BatchCounters{}, bc.NewError("Failed to query watermark offsets: %v", err) } + if !created && bc.mode != "retry" && hasLag(commitedOffset, highOffset) { + //Self-heal (JITSU-214). Topics are sharded, so this consumer is the + //group's only member for its partition: if it has been alive long enough + //to have joined and still owns nothing while messages are waiting, it + //has dropped out of the group (max.poll.interval exceeded, fenced by a + //restart, or the empty group was garbage-collected by the broker) and + //would otherwise poll nothing every period forever, looking idle. + if reason := bc.assignmentLost(consumer); reason != "" { + bc.membershipLost(reason) + bc.restartConsumer(nil) + consumer = bc.consumer.Load() + if consumer == nil { + bc.errorMetric("resume_error") + return BatchCounters{}, bc.NewError("Failed to recreate kafka consumer after losing group membership") + } + } + } if !bc.shouldConsume(partition, commitedOffset, highOffset) { bc.Debugf("Consumer should not consume. offsets: %d-%d", commitedOffset, highOffset) return BatchCounters{}, nil @@ -380,12 +443,46 @@ func (bc *AbstractBatchConsumer) close() error { } consumer := bc.consumer.Swap(nil) if consumer != nil { - err := consumer.Close() - return err + return bc.closeConsumer(consumer) } return nil } +// closeConsumer closes a kafka consumer exactly once, however many paths hold +// its pointer (see closedConsumers). Returns the Close error, or nil when the +// consumer was already closed by another path. +func (bc *AbstractBatchConsumer) closeConsumer(consumer *kafka.Consumer) error { + bc.closedMu.Lock() + if _, done := bc.closedConsumers[consumer]; done { + bc.closedMu.Unlock() + return nil + } + bc.closedConsumers[consumer] = struct{}{} + bc.closedMu.Unlock() + e1 := consumer.Unsubscribe() + e2 := consumer.Close() + bc.Infof("Consumer closed: %s unsubscribe: %v close: %v", consumer.String(), e1, e2) + return e2 +} + +// quarantineClose closes a consumer that has just been replaced, after a delay +// long enough for anything that loaded the old pointer before the swap — the +// paused heartbeat mid-ReadMessage, most importantly — to return from it. +// librdkafka is not safe against a poll racing a Close on the same instance. +// +// The delay also bounds how long a replaced static member stays registered: +// it is deregistered a session timeout after this close, and only then is its +// base group.instance.id free again. Keep +// BATCH_RUNNER_WAIT_FOR_MESSAGES_SEC + 5s + KAFKA_SESSION_TIMEOUT_MS under the +// 60s minimum gap that pauseOrSuspend requires before suspending, so a +// suspended consumer recreated with the base id never collides with it. +func (bc *AbstractBatchConsumer) quarantineClose(consumer *kafka.Consumer) { + safego.RunWithRestart(func() { + time.Sleep(bc.waitForMessages + 5*time.Second) + _ = bc.closeConsumer(consumer) + }) +} + func (bc *AbstractBatchConsumer) processBatch(destination *Destination, batchNum, batchSize, batchSizeBytes, retryBatchSize int, highOffset int64, updatedHighOffset int) (counters BatchCounters, state bulker.State, nextBath bool, err error) { bc.resume() return bc.batchFunc(destination, batchNum, batchSize, batchSizeBytes, retryBatchSize, highOffset, updatedHighOffset) @@ -418,8 +515,14 @@ func (bc *AbstractBatchConsumer) pauseOrSuspend(startedAt time.Time) { if bc.config.SuspendConsumers && timeToNextBatch >= 60*time.Second { bc.Infof("Suspending consumer %s for %s", consumer.String(), timeToNextBatch) bc._unpause() - _ = consumer.Close() - bc.consumer.Store(nil) + //only suspend the consumer we looked at: a restart may have swapped in + //a new one meanwhile, and storing nil over it would leak a live group + //member. That one stays up and is used by the next period. + if bc.consumer.CompareAndSwap(consumer, nil) { + _ = bc.closeConsumer(consumer) + } else { + bc.Infof("Consumer was replaced while suspending. Keeping the new one") + } } else { bc.pause(false) } @@ -442,7 +545,7 @@ func (bc *AbstractBatchConsumer) pause(immediatePoll bool) { errorReported := false firstPoll := immediatePoll //this loop keeps heatbeating consumer to prevent it from being kicked out from group - pauseTicker := time.NewTicker(time.Duration(bc.config.KafkaMaxPollIntervalMs) * time.Millisecond / 2) + pauseTicker := time.NewTicker(bc.heartbeatInterval()) defer pauseTicker.Stop() loop: for { @@ -497,13 +600,18 @@ func (bc *AbstractBatchConsumer) pause(immediatePoll bool) { bc.Errorf("Error on paused consumer: %v", kafkaErr) errorReported = true } - if kafkaErr.IsRetriable() { + if reason := membershipLossReason(kafkaErr); reason != "" { + //left the group: waiting does not bring it back (that is the + //zombie this fix is about), so rejoin with a new consumer + bc.membershipLost(reason) + bc.restartConsumerAsync() + } else if kafkaErr.IsRetriable() && !kafkaErr.IsFatal() { time.Sleep(10 * time.Second) } else { - //restartConsumer blocks for KafkaSessionTimeoutMs + 15s - //baseline per init attempt; running it synchronously - //here would starve the resumeChannel select and trip - //"Resume timeout" in resume() once 5 min elapses. + //restartConsumer can block for KafkaSessionTimeoutMs + 15s per + //failed creation attempt (broker unreachable); running it + //synchronously here would starve the resumeChannel select and + //trip "Resume timeout" in resume() once 5 min elapses. bc.restartConsumerAsync() } } else if message != nil { @@ -521,31 +629,120 @@ func (bc *AbstractBatchConsumer) pause(immediatePoll bool) { }) } -func (bc *AbstractBatchConsumer) initConsumer(force bool) (consumer *kafka.Consumer, err error) { +// initConsumer returns the current kafka consumer, creating one when there is +// none. `created` reports whether this call made it: a new consumer has not +// polled yet, so it has no assignment and no group membership to check. +func (bc *AbstractBatchConsumer) initConsumer() (consumer *kafka.Consumer, created bool, err error) { consumer = bc.consumer.Load() - if consumer == nil || force { - consumer, err = kafka.NewConsumer(&bc.consumerConfig) - if err != nil { - bc.errorMetric("consumer_error:" + metrics.KafkaErrorCode(err)) - bc.Errorf("Error creating kafka consumer: %v", err) - return nil, err - } - err = consumer.SubscribeTopics([]string{bc.topicId}, bc.rebalanceCallback) - if err != nil { - bc.errorMetric("consumer_error:" + metrics.KafkaErrorCode(err)) - _ = consumer.Close() - bc.Errorf("Failed to subscribe to topic: %v", err) - return nil, err - } - //if bc.mode == "retry" && bc.topicId == bc.config.KafkaDestinationsRetryTopicName { - // consumer.Assign([]kafka.TopicPartition{kafka.TopicPartition{Topic: &bc.topicId, Offset: kafka.OffsetStored, Partition: int32(bc.config.InstanceIndex)}}) - //} - bc.Infof("Consumer created: %s", consumer.String()) - bc.consumer.Store(consumer) + if consumer != nil { + return consumer, false, nil } + consumer, err = bc.newConsumer(false) + if err != nil { + return nil, false, err + } + //this consumer replaces nothing (first run, or a start after a suspend), so + //it is not covered by the restart cooldown — clear any stamp an earlier + //restart left behind, or a fatal error on it would be skipped + bc.lastRestartAt.Store(0) + bc.consumer.Store(consumer) + return consumer, true, nil +} + +// newConsumer creates and subscribes a kafka consumer without publishing it. +// +// A restart gets a "-rN" suffix on the static group.instance.id, because the +// consumer it replaces is still a live group member for up to a session +// timeout and two members sharing an instance id fence each other. Any other +// creation — the first one, or one after a suspend, where the previous member +// is long gone — keeps the base id, so a pod restart resumes as the same +// static member and the partition assignment order across instances holds. +func (bc *AbstractBatchConsumer) newConsumer(restart bool) (*kafka.Consumer, error) { + config := kafka.ConfigMap(utils.MapPutAll(kafka.ConfigMap{}, bc.consumerConfig)) + instanceId := bc.GetInstanceId() + if restart { + instanceId = fmt.Sprintf("%s-r%d", instanceId, bc.restartGeneration.Add(1)) + config["group.instance.id"] = instanceId + } + consumer, err := kafka.NewConsumer(&config) + if err != nil { + bc.errorMetric("consumer_error:" + metrics.KafkaErrorCode(err)) + bc.Errorf("Error creating kafka consumer: %v", err) + return nil, err + } + err = consumer.SubscribeTopics([]string{bc.topicId}, bc.rebalanceCallback) + if err != nil { + bc.errorMetric("consumer_error:" + metrics.KafkaErrorCode(err)) + _ = consumer.Close() + bc.Errorf("Failed to subscribe to topic: %v", err) + return nil, err + } + //if bc.mode == "retry" && bc.topicId == bc.config.KafkaDestinationsRetryTopicName { + // consumer.Assign([]kafka.TopicPartition{kafka.TopicPartition{Topic: &bc.topicId, Offset: kafka.OffsetStored, Partition: int32(bc.config.InstanceIndex)}}) + //} + bc.consumerCreatedAt.Store(time.Now().UnixNano()) + bc.Infof("Consumer created: %s (group.instance.id: %s)", consumer.String(), instanceId) return consumer, nil } +// heartbeatInterval is how often a paused consumer is polled to stay in the +// group. It must leave a real margin under max.poll.interval.ms: at exactly +// half, one poll that waits its full timeout plus any scheduling delay is +// enough to overshoot, and librdkafka then leaves the group (JITSU-214). +func (bc *AbstractBatchConsumer) heartbeatInterval() time.Duration { + return time.Duration(bc.config.KafkaMaxPollIntervalMs) * time.Millisecond / 3 +} + +// membershipGrace is how long a consumer gets to join the group before an +// empty assignment counts as lost membership: the paused heartbeat first +// polls it after one heartbeatInterval, and joining takes up to a session. +func (bc *AbstractBatchConsumer) membershipGrace() time.Duration { + return bc.heartbeatInterval() + time.Duration(bc.config.KafkaSessionTimeoutMs)*time.Millisecond +} + +// assignmentLost reports why the current consumer no longer looks like a group +// member, or "" while it does. Only meaningful once the consumer is past the +// grace period — before that an empty assignment is just "not joined yet". +func (bc *AbstractBatchConsumer) assignmentLost(consumer *kafka.Consumer) string { + age := time.Since(time.Unix(0, bc.consumerCreatedAt.Load())) + if age < bc.membershipGrace() { + return "" + } + partitions, err := consumer.Assignment() + if err != nil { + return fmt.Sprintf("assignment query failed after %s: %v", age.Round(time.Second), err) + } + if len(partitions) == 0 { + return fmt.Sprintf("no partition assignment after %s while messages are waiting", age.Round(time.Second)) + } + return "" +} + +// membershipLost records that this consumer dropped out of its group — the +// condition behind topics that silently stop being consumed (JITSU-214). It is +// an error-level log and a dedicated metric label so it can be alerted on. +// Every caller follows it with a restart, which logs its own progress. +func (bc *AbstractBatchConsumer) membershipLost(reason string) { + bc.errorMetric("membership_lost") + bc.Errorf("Consumer group membership lost: %s", reason) +} + +// onReadError handles a non-timeout error from ReadMessage during batch +// processing. A fatal error leaves the librdkafka instance permanently +// inoperable and a non-retriable one is not going to clear on its own; either +// way every later poll on this object just times out and the topic looks idle, +// so the consumer is recreated (asynchronously — the caller is inside a batch +// and holds the consumer lock). +func (bc *AbstractBatchConsumer) onReadError(kafkaErr kafka.Error) { + bc.errorMetric("consumer_error:" + metrics.KafkaErrorCode(kafkaErr)) + if reason := membershipLossReason(kafkaErr); reason != "" { + bc.membershipLost(reason) + } + if kafkaErr.IsFatal() || !kafkaErr.IsRetriable() { + bc.restartConsumerAsync() + } +} + // restartConsumerAsync schedules restartConsumer to run in a separate // goroutine, returning immediately. Re-entry is suppressed: if a restart // is already in flight, the call is a no-op. Use from contexts that must @@ -562,44 +759,112 @@ func (bc *AbstractBatchConsumer) restartConsumerAsync() { }() } +// restartConsumer replaces the kafka consumer with a fresh one. +// +// The new consumer is created and published FIRST; the old one is closed +// afterwards, in quarantine (see quarantineClose). Two things make that safe: +// - the new consumer joins under a different group.instance.id (see +// newConsumer), so the broker never fences one static member with the +// other — the FENCED_INSTANCE_ID fatal error that killed consumers when +// an old member was still alive while its replacement joined (JITSU-214). +// The old member's assignment is released by the broker once its session +// times out, at which point the new one is assigned the partition; +// - nothing ever closes a consumer another goroutine may still be polling: +// the pointer swap happens before the close, and the close waits out any +// in-flight poll. +// +// Restarts are serialized, and a consumer younger than a session timeout is +// not restarted again: whoever asked for it was looking at the previous one. func (bc *AbstractBatchConsumer) restartConsumer(beforeInit func()) { if bc.retired.Load() { return } + bc.restartMu.Lock() + defer bc.restartMu.Unlock() + if bc.retired.Load() { + return + } + sessionTimeout := time.Duration(bc.config.KafkaSessionTimeoutMs) * time.Millisecond + //Only a consumer that a restart installed is subject to the cooldown: it + //means a concurrent restart already replaced the one this caller was + //looking at. A consumer created by initConsumer (the first one, or one + //started after a suspend) is nobody's replacement, so a fatal error on it + //must still force a restart rather than leave an unusable handle in place. + lastRestart := bc.lastRestartAt.Load() + if lastRestart > 0 && bc.consumer.Load() != nil && time.Since(time.Unix(0, lastRestart)) < sessionTimeout { + bc.Infof("Consumer was restarted less than %s ago. Skipping restart", sessionTimeout) + //beforeInit still runs: it is the caller's own recovery work (an offset + //fix-up via the admin client), and skipping it because someone else + //restarted the consumer would silently drop that repair + if beforeInit != nil { + beforeInit() + } + return + } bc.Infof("Restarting consumer") - go func(c *kafka.Consumer) { - e1 := c.Unsubscribe() - e2 := c.Close() - bc.Infof("Previous consumer closed: %v %v", e1, e2) - }(bc.consumer.Load()) - - ticker := time.NewTicker(time.Duration(bc.config.KafkaSessionTimeoutMs+15000) * time.Millisecond) - defer ticker.Stop() // for faster reaction on retiring pauseTicker := time.NewTicker(1 * time.Second) defer pauseTicker.Stop() - + retry := time.NewTicker(sessionTimeout + 15*time.Second) + defer retry.Stop() for { - select { - case <-pauseTicker.C: - if bc.idle.Load() && bc.retired.Load() { + //beforeInit runs before every attempt, as it always did: callers pass + //idempotent work (an offset fix-up via the admin client) that must run + //right before the consumer that will use its result + if beforeInit != nil { + beforeInit() + } + consumer, err := bc.newConsumer(true) + if err == nil { + //creating a consumer takes seconds; the object may have been retired + //and closed meanwhile. Publishing now would leave a live, subscribed + //group member nobody ever closes. + if bc.retired.Load() && bc.idle.Load() { + bc.Infof("Consumer was retired while restarting. Discarding the new consumer") + _ = bc.closeConsumer(consumer) return } - case <-ticker.C: - if beforeInit != nil { - beforeInit() + bc.lastRestartAt.Store(time.Now().UnixNano()) + if old := bc.consumer.Swap(consumer); old != nil { + bc.quarantineClose(old) } - _, err := bc.initConsumer(true) - if err != nil { - break + //Retirement can also land between the check above and this swap, + //with close() running in between: it would have taken the old + //consumer out and closed it, and nothing would ever close the one + //just published. Withdraw it — unless someone already took it out, + //in which case they close it. + if bc.retired.Load() && bc.idle.Load() && bc.consumer.CompareAndSwap(consumer, nil) { + bc.Infof("Consumer was retired while restarting. Discarding the new consumer") + _ = bc.closeConsumer(consumer) } return } + //creation failed (broker unreachable?): retry after a session timeout, + //polling for retirement in between. Retirement alone ends the loop — + //this runs synchronously from ConsumeAll too, where idle is false for + //the whole run, so waiting for idle would never let a retired consumer + //out (and it holds bc.Mutex and restartMu while it waits). + for { + select { + case <-pauseTicker.C: + if bc.retired.Load() { + return + } + continue + case <-retry.C: + } + break + } } } func (bc *AbstractBatchConsumer) pauseKafkaConsumer() { consumer := bc.consumer.Load() + if consumer == nil { + //a restart is replacing the consumer; rebalanceCallback pauses the new + //one on assignment while paused is set + return + } partitions, err := consumer.Assignment() if len(partitions) > 0 { err = consumer.Pause(partitions) @@ -664,6 +929,10 @@ func (bc *AbstractBatchConsumer) resume() { bc.SystemErrorf("failed to resume kafka consumer.: %v", err) } }() + if consumer == nil { + err = bc.NewError("no kafka consumer (restart in progress?)") + return + } partitions, err := consumer.Assignment() if err != nil { return @@ -723,6 +992,36 @@ func (bc *AbstractBatchConsumer) countersMetric(counters BatchCounters) { } } +// hasLag reports whether the topic holds messages this consumer has not +// committed: either nothing was ever committed (OffsetBeginning/invalid) or the +// committed offset trails the high watermark. +func hasLag(committedOffset, highOffset int64) bool { + if highOffset <= 0 { + return false + } + return committedOffset < 0 || committedOffset < highOffset +} + +// membershipLossReason maps a kafka error to a human-readable reason when it +// means the consumer is no longer a member of its group, or "" otherwise. +// These are the ways a static member drops out: it stopped polling in time +// (librdkafka leaves the group itself), a newer instance with the same +// group.instance.id took its place, or the coordinator forgot it. +func membershipLossReason(kafkaErr kafka.Error) string { + switch kafkaErr.Code() { + case kafka.ErrMaxPollExceeded: + return "max.poll.interval.ms exceeded, librdkafka left the group" + case kafka.ErrFencedInstanceID, kafka.ErrFenced: + return "fenced by another consumer with the same group.instance.id" + case kafka.ErrUnknownMemberID: + return "coordinator no longer knows this member" + } + if kafkaErr.IsFatal() { + return fmt.Sprintf("fatal consumer error: %v", kafkaErr) + } + return "" +} + type BatchCounters struct { consumed int skipped int diff --git a/bulker/bulkerapp/app/abstract_batch_consumer_test.go b/bulker/bulkerapp/app/abstract_batch_consumer_test.go new file mode 100644 index 000000000..7ca2531f2 --- /dev/null +++ b/bulker/bulkerapp/app/abstract_batch_consumer_test.go @@ -0,0 +1,54 @@ +package app + +import ( + "testing" + + "github.com/confluentinc/confluent-kafka-go/v2/kafka" + "github.com/stretchr/testify/assert" +) + +func TestHasLag(t *testing.T) { + testCases := []struct { + desc string + committed int64 + high int64 + expected bool + }{ + {"empty topic, nothing committed", int64(kafka.OffsetBeginning), 0, false}, + {"empty topic, stale commit", 10, 0, false}, + {"messages but no committed offset (new topic or deleted group)", int64(kafka.OffsetBeginning), 262_072, true}, + {"committed behind the watermark", 100, 150, true}, + {"fully consumed", 150, 150, false}, + } + for _, tc := range testCases { + t.Run(tc.desc, func(t *testing.T) { + assert.Equal(t, tc.expected, hasLag(tc.committed, tc.high)) + }) + } +} + +func TestMembershipLossReason(t *testing.T) { + testCases := []struct { + desc string + err kafka.Error + lost bool + contains string + }{ + {"max.poll.interval exceeded", kafka.NewError(kafka.ErrMaxPollExceeded, "Application maximum poll interval (300000ms) exceeded", false), true, "max.poll.interval"}, + {"fenced static instance (broker)", kafka.NewError(kafka.ErrFencedInstanceID, "Static consumer fenced by other consumer with same group.instance.id", true), true, "fenced"}, + {"fenced (local)", kafka.NewError(kafka.ErrFenced, "fenced", true), true, "fenced"}, + {"unknown member", kafka.NewError(kafka.ErrUnknownMemberID, "Unknown member", false), true, "member"}, + {"any fatal error", kafka.NewError(kafka.ErrFatal, "fatal", true), true, "fatal"}, + {"timeout is not a membership loss", kafka.NewError(kafka.ErrTimedOut, "timed out", false), false, ""}, + {"transient broker error is not a membership loss", kafka.NewError(kafka.ErrTransport, "transport", false), false, ""}, + } + for _, tc := range testCases { + t.Run(tc.desc, func(t *testing.T) { + reason := membershipLossReason(tc.err) + assert.Equal(t, tc.lost, reason != "") + if tc.contains != "" { + assert.Contains(t, reason, tc.contains) + } + }) + } +} diff --git a/bulker/bulkerapp/app/batch_consumer.go b/bulker/bulkerapp/app/batch_consumer.go index a493d6dfd..50407e9c7 100644 --- a/bulker/bulkerapp/app/batch_consumer.go +++ b/bulker/bulkerapp/app/batch_consumer.go @@ -149,7 +149,7 @@ func (bc *BatchConsumerImpl) processBatchImpl(destination *Destination, batchNum // waitForMessages period is over. it's ok. considering batch as full break } - bc.errorMetric("consumer_error:" + metrics.KafkaErrorCode(kafkaErr)) + bc.onReadError(kafkaErr) if bulkerStream != nil { _ = bulkerStream.Abort(ctx) } @@ -224,7 +224,7 @@ func (bc *BatchConsumerImpl) processBatchImpl(destination *Destination, batchNum if processed == batchSize { nextBatch = true } - pauseTimer := time.AfterFunc(time.Duration(bc.config.KafkaMaxPollIntervalMs)*time.Millisecond/2, func() { + pauseTimer := time.AfterFunc(bc.heartbeatInterval(), func() { // we need to pause consumer to avoid kafka session timeout while loading huge batches to slow destinations bc.pause(true) }) @@ -270,7 +270,16 @@ func (bc *BatchConsumerImpl) processBatchImpl(destination *Destination, batchNum } counters.processed = processed counters.processedBytes = consumedBytes - _, err = consumer.CommitMessage(latestMessage) + //re-read the pointer: loading a batch into a slow destination can outlast + //a consumer restart (the pause heartbeat may have replaced the consumer + //this batch was read with), and committing on the retired handle would + //fail after it is closed + consumer = bc.consumer.Load() + if consumer == nil { + err = bc.NewError("no kafka consumer to commit the batch with") + } else { + _, err = consumer.CommitMessage(latestMessage) + } if err != nil { bc.errorMetric("KAFKA_COMMIT_ERR:" + metrics.KafkaErrorCode(err)) bc.Errorf("Failed to commit kafka consumer after batch was successfully committed to the destination: %v", err) @@ -299,7 +308,13 @@ func (bc *BatchConsumerImpl) processBatchImpl(destination *Destination, batchNum }) if !committed { var tp []kafka.TopicPartition - tp, err = bc.consumer.Load().CommitMessage(latestMessage) + //the restart above may have left no consumer at all (retired + //meanwhile), and the offset repair did not run either + if current := bc.consumer.Load(); current == nil { + err = bc.NewError("no kafka consumer to commit the batch with after restart") + } else { + tp, err = current.CommitMessage(latestMessage) + } if err != nil { bc.SystemErrorf("Failed to commit kafka consumer after batch was successfully committed to the destination: %v", err) err = bc.NewError("Failed to commit kafka consumer: %v", err) diff --git a/bulker/bulkerapp/app/retry_consumer.go b/bulker/bulkerapp/app/retry_consumer.go index 35bd53e4f..f9686e5d6 100644 --- a/bulker/bulkerapp/app/retry_consumer.go +++ b/bulker/bulkerapp/app/retry_consumer.go @@ -139,6 +139,7 @@ func (rc *RetryConsumer) processBatchImpl(_ *Destination, _, _, _, retryBatchSiz // waitForMessages period is over. it's ok. considering batch as full break } + rc.onReadError(kafkaErr) return counters, state, false, rc.NewError("Failed to consume event from topic. Retryable: %t: %v", kafkaErr.IsRetriable(), kafkaErr) } if firstPosition != nil && message.TopicPartition.Partition != firstPosition.Partition {