chore: added-logs-handled-infinite-loop

Author: saksham-datazip

drivers/kafka/internal/cdc.go

@@ -183,6 +183,8 @@ func (k *Kafka) PostCDC(ctx context.Context, readerIdx int) error {
 			if reader == nil {
 				return fmt.Errorf("reader %s not found for commit", readerID)
 			}
+			_, generationID := reader.GroupMetadata()
+			logger.Debugf("reader %s post cdc: generation id: %d", readerID, generationID)
 
 			if err := reader.CommitRecords(ctx, messages...); err != nil {
 				return fmt.Errorf("commit failed for reader %s: %s", readerID, err)
@@ -208,25 +210,28 @@ func (k *Kafka) PostCDC(ctx context.Context, readerIdx int) error {
 // for processing messages from a Kafka reader.
 func (k *Kafka) processKafkaMessages(ctx context.Context, reader *kgo.Client, stopProcessFn func(record types.KafkaRecord) (bool, error)) error {
 	for {
-		messages := reader.PollFetches(ctx)
+		pollCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
+		messages := reader.PollFetches(pollCtx)
+		pollCtxErr := pollCtx.Err()
+		cancel()
 		errs := messages.Errors()
-		if len(errs) > 0 {
+
+		// Rebalance/exit checks must run even when PollFetches returns an empty batch;
+		// otherwise a reader reassigned to empty partitions never enters the record loop.
+		if stopProcessing, err := k.readerManager.FetchExitState(); stopProcessing {
+			return err
+		}
+
+		if len(errs) > 0 && pollCtxErr == nil {
 			return fmt.Errorf("%v: error reading message in Kafka CDC sync: %v", constants.ErrNonRetryable, errs[0].Err)
 		}
 
 		records := messages.RecordIter()
 
 		for !records.Done() {
-			// Discover/schema sampling uses standalone partition consumers without ReaderManager.
-			if k.readerManager != nil {
-				if k.readerManager.ShouldStopProcessing() {
-					logger.Infof("stopping kafka CDC processing due to consumer group rebalance")
-					return nil
-				}
-				if err := k.readerManager.ErrForExitMode(); err != nil {
-					logger.Errorf("kafka consumer lost partition ownership and CDC processing can no longer continue safely: %s	", err)
-					return err
-				}
+			// Re-check rebalance state before processing each record to stop immediately after partition revocation.
+			if stopProcessing, err := k.readerManager.FetchExitState(); stopProcessing {
+				return err
 			}
 			message := records.Next()
 

pkg/kafka/reader.go

@@ -361,21 +361,24 @@ func (r *ReaderManager) RebalanceDetected(client *kgo.Client) bool {
 	return generationID >= 0 && generationID != r.generationID.Load()
 }
 
-// ShouldStopProcessing reports a consumer-group rebalance (assign/revoke during CDC).
-// The fetch loop must exit with nil — not an error — so abstract layer does not retry.
-func (r *ReaderManager) ShouldStopProcessing() bool {
-	return r.exitMode.Load() == gracefulExit
-}
+// FetchExitState reports whether CDC processing should stop after PollFetches.
+// exitMode is updated by consumer group rebalance callbacks before PollFetches returns.
+func (r *ReaderManager) FetchExitState() (stop bool, err error) {
+	// ReaderManager will be nil during discover mode.
+	if r == nil {
+		return false, nil
+	}
 
-// ErrForExitMode returns an error only for nonRetryableExit (e.g. partitions lost).
-func (r *ReaderManager) ErrForExitMode() error {
 	switch r.exitMode.Load() {
-	case normalProcessing, gracefulExit:
-		return nil
+	case normalProcessing:
+		return false, nil
+	case gracefulExit:
+		logger.Infof("stopping kafka CDC processing gracefully due to consumer group rebalance")
+		return true, nil
 	case nonRetryableExit:
-		return fmt.Errorf("%v: kafka sync aborted", constants.ErrNonRetryable)
+		return true, fmt.Errorf("%v: kafka sync aborted due to partition loss during consumer group rebalance", constants.ErrNonRetryable)
 	default:
-		return fmt.Errorf("%v: kafka sync aborted", constants.ErrNonRetryable)
+		return true, fmt.Errorf("%v: kafka sync aborted: unexpected exit mode", constants.ErrNonRetryable)
 	}
 }