High memory usage / input continues consuming when output is unavailable (back-pressure not applied aggressively enough)

When the output becomes unavailable (connection lost, broker down, network partition, etc.), the input continues to consume messages from Kafka/Redpanda and holds them in memory. This leads to significant memory growth and can eventually cause OOM.

According to the documentation, outputs should apply back-pressure and "gracefully stop consuming", but with the following configuration the back-pressure is not aggressive enough and memory keeps increasing.

Expected behavior

When the output is blocked/unavailable for a prolonged time, the input should stop (or heavily slow down) consuming new messages so that memory usage stays bounded.

Actual behavior

• Input continues fetching messages
• Messages accumulate in internal buffers (partition_buffer_bytes, pipeline, retry, etc.)
• Memory usage grows continuously until the process is killed or the output recovers

Configuration (relevant parts)

input:
  label: kafka_in
  batched:
    child:
      redpanda:
        seed_brokers:
          - ${KAFKA_BROKERS}
        topics:
          - ${KAFKA_TOPIC}
        consumer_group: ${KAFKA_GROUP}
        start_offset: earliest
        commit_period: 100ms
        auto_replay_nacks: true
        fetch_min_bytes: 1MiB
        fetch_max_wait: 50ms
        fetch_max_bytes: 16MiB
        fetch_max_partition_bytes: 2MiB
        partition_buffer_bytes: 4MiB
        # ... other kafka settings
    policy:
      count: ${BATCH_COUNT}
      byte_size: ${BATCH_BYTE_SIZE}
      period: ${BATCH_PERIOD}

output:
  label: kafka_out
  processors:
    - rate_limit:
        resource: output_rate_limiter
  retry:
    max_retries: ${RETRY_MAX_RETRIES}
    backoff:
      initial_interval: ${RETRY_BACKOFF_INITIAL}
      max_interval: ${RETRY_BACKOFF_MAX}
      max_elapsed_time: 0s          # infinite retry
    output:
      redpanda:
        seed_brokers:
          - ${KAFKA_EXTERNAL_BROKERS}
        topic: ${KAFKA_EXTERNAL_TOPIC}
        # ... sasl, tls, batching, etc.

Start by reproducing the configuration in the input and output sections with the output unavailable and infinite retry enabled. Observe consumption, internal buffering, and memory growth while the output is blocked. Done means consumption stops or slows enough to keep memory bounded, with behavior verified during output recovery.