fix(kafka): keep group consumer polling while Finish waits on deferred acks

internal/pkg/pipeline/task/kafka/README.md

@@ -235,7 +235,7 @@ See `test/pipelines/kafka_acking.yaml` for a fixture that exercises deferred com
  - **Out-of-range stored offset:** librdkafka logs a `%4|OFFSET ... offset reset` warning when this happens. It's informational — the consumer self-recovers to the position implied by `auto_offset_reset`. The warning persists across restarts until a successful read commits a new valid offset (or until you manually reset the group offsets at the broker).
  - **`end_after`** sets a wall-clock read deadline distinct from `retry_limit` (which is idle-based). Use `end_after` when you want a guaranteed stop time even on a busy topic. Worst-case shutdown latency is one `timeout` window because in-flight `ReadMessage` polls cannot be canceled mid-flight.
  - **`max_records`** is count-based and independent of `end_after`/`retry_limit`. The counter increments after each record is forwarded downstream, so the cap is exact for delivered records. If the topic has fewer than `max_records` available, the reader keeps polling until `retry_limit` or `end_after` fires.
- - **Group commits** use Kafka auto-commit every 5000ms. Auto offset store is disabled, so offsets are stored only after downstream completion (see [Message acknowledgment](#message-acknowledgment-group-read-mode) above). Write mode is unaffected: the producer settles each record on its delivery report, so an upstream source that defers acknowledgment is held until the broker confirms the write.
+ - **Group commits** use Kafka auto-commit every 5000ms. Auto offset store is disabled, so offsets are stored only after downstream completion (see [Message acknowledgment](#message-acknowledgment-group-read-mode) above). After the reader stops (`max_records`, `end_after`, idle retry limit, or all partitions paused), `Finish` keeps polling until those acks settle so the consumer stays in the group — `max.poll.interval.ms` (librdkafka default 5m) is an application-poll deadline, not covered by `heartbeat.interval.ms`. Write mode is unaffected: the producer settles each record on its delivery report, so an upstream source that defers acknowledgment is held until the broker confirms the write.
  - **Read isolation** is set to `read_committed` for both standalone and group consumers — this is the consumer-side complement to `idempotent: true` on the producer and ensures consumers never read uncommitted or aborted messages.
  - The init broker probe always uses the 15s default timeout regardless of the configured `timeout` to allow for SCRAM+TLS handshake round trips.
  - **Message format** defaults to `json` (raw bytes pass through). Set `format: avro` to enable Confluent Avro serialization; this requires `schema_registry_url` and a pre-registered schema. The `schema_registry_url` field alone does **not** activate Avro — `format: avro` must be set explicitly.
@@ -247,6 +247,8 @@ See `test/pipelines/kafka_acking.yaml` for a fixture that exercises deferred com
 - If TLS connections fail, verify the CA at `cert_path` or `cert` matches the broker's certificate chain. Also check whether the certificate at `cert` is correctly formatted (PEM) in multiline YAML (use `|` and indentation).
 - If SASL/SCRAM authentication fails, double-check `username`/`password` and the broker's configured mechanism.
 
+- `MAXPOLL` / `Application maximum poll interval exceeded` followed by `Broker: Unknown member` on commit means the consumer sat without polling longer than `max.poll.interval.ms` while waiting for downstream tasks to settle.
+
 - If standalone reads fail with a group authorization error, ask your Kafka admin to run:
   ```
   kafka-acls.sh --add --allow-principal User:<principal> \

internal/pkg/pipeline/task/kafka/kafka.go

@@ -297,6 +297,14 @@ func (k *kafka) read(ctx context.Context, output chan<- *record.Record) error {
 	if err != nil {
 		return err
 	}
+	if r.group {
+		// Pause assigned partitions so Finish heartbeat polls do not fetch new records.
+		defer func() {
+			if err := r.pauseAll(); err != nil {
+				fmt.Printf("warning: failed to pause partitions for topic %s: %v\n", k.Topic, err)
+			}
+		}()
+	}
 
 	codec, err := k.newCodec()
 	if err != nil {

internal/pkg/pipeline/task/kafka/reader.go

@@ -69,12 +69,66 @@ func (k *kafka) registerReader(r *reader) {
 	k.readers = append(k.readers, r)
 }
 
+const heartbeatPollMs = 1000
+
+// Group consumers must keep polling after Read stops: max.poll.interval.ms (5m)
+// is an application-poll deadline that heartbeat.interval.ms does not reset.
+func waitWhile(wait func(), poll func()) {
+	done := make(chan struct{})
+	go func() {
+		wait()
+		close(done)
+	}()
+	for {
+		select {
+		case <-done:
+			return
+		default:
+			poll()
+		}
+	}
+}
+
+func (k *kafka) heartbeatPoll() {
+	k.readersMu.Lock()
+	readers := slices.Clone(k.readers)
+	k.readersMu.Unlock()
+	for _, r := range readers {
+		r.pollHeartbeat()
+	}
+}
+
+func (r *reader) pollHeartbeat() {
+	if r.consumer == nil || !r.group {
+		return
+	}
+	r.consumerMu.Lock()
+	defer r.consumerMu.Unlock()
+	ev := r.consumer.Poll(heartbeatPollMs)
+	if err, ok := ev.(ckafka.Error); ok && err.Code() != ckafka.ErrTimedOut {
+		fmt.Printf("warning: kafka heartbeat poll for topic %s: %v\n", r.k.Topic, err)
+	}
+}
+
+func (r *reader) pauseAll() error {
+	r.consumerMu.Lock()
+	defer r.consumerMu.Unlock()
+	assignment, err := r.consumer.Assignment()
+	if err != nil {
+		return err
+	}
+	if len(assignment) == 0 {
+		return nil
+	}
+	return r.consumer.Pause(assignment)
+}
+
 // Finish, not Run, waits for deferred stores: a join that emits on input close
 // cannot settle until this task's output channel is closed, which happens only
 // after Run returns.
 func (k *kafka) Finish() error {
 	if k.tracker != nil {
-		k.tracker.Wait()
+		waitWhile(k.tracker.Wait, k.heartbeatPoll)
 	}
 
 	k.readersMu.Lock()