kafka_franz input+output: ":4195/ready" is not working correctly

[marcluer]

When the broker (e.g. redpanda:29092) is down, when Connect is started, "localhost:4195/ready" works as expected:

{"statuses":[
    {"label":"","path":"input","connected":false,"error":"failed to connect to cluster: unable to dial: dial tcp: lookup redpanda on 127.0.0.11:53: no such host"},
    {"label":"","path":"output","connected":true}]}

Once the broker is up, and everything works, "localhost:4195/ready" is working as well:

{"statuses":[
    {"label":"","path":"input","connected":true},
    {"label":"","path":"output","connected":true}]}

When now the broker becomes unavailable, "localhost:4195/ready" does not reflect that and still shows:

{"statuses":[
    {"label":"","path":"input","connected":true},
    {"label":"","path":"output","connected":true}]}

This issue seems to affect kafka_franz input + output. When using kafka instead of kafka_franz, atleast the input state is correctly represented by "/ready". But kafka output also seems to be wrong. (more testing required, though)

pipeline.yml

logger:
  level: INFO
  format: json

input:
  kafka_franz:
    seed_brokers:
      - redpanda:29092
    topics:
      - 'global'
    consumer_group: "postgres"

pipeline:
  processors:
    - log:
        level: INFO
        message: ""
        fields_mapping: |
          root.meta.ts = (meta("kafka_timestamp_ms").number() / 1000).ts_format("2006-01-02 15:04:05.999999+00")
          root.meta.kafka_key = meta("kafka_key")
          root.meta.content = content()