apache / apache/pulsar-connectors
[improve][kafka-connect-adaptor] An unproductive source is indistinguishable from an idle one
- Dominant language
- Java
- Stars
- 26
- Forks
- 25
- PR merge metrics
- No merged PRs in 30d
Description
## What happens
A source built on `kafka-connect-adaptor` whose wrapped Kafka Connect task can never
produce a record — wrong host, unreachable database, bad credentials — reports itself as
perfectly healthy for as long as it runs. `pulsar-admin sources status` shows:
```json
{"running": true, "error": "", "numRestarts": 0,
"numReceivedFromSource": 0, "lastReceivedTime": 0}
```
There is no restart, no exception, and no error string. Operationally this is
indistinguishable from a correctly configured source that simply has no traffic yet.
## Why
`AbstractKafkaConnectSource.read()` discards the fact that a poll produced nothing:
```java
public synchronized Record read() throws Exception {
while (true) {
if (currentBatch == null) {
List recordList = sourceTask.poll();
if (recordList == null || recordList.isEmpty()) {
continue; // <-- nothing counted, timed or logged
}
```
Nothing counts consecutive empty polls or timestamps the last non-empty one, so there is
no state for the runtime to surface. `lastReceivedTime` stays `0`, which is the same value
a genuinely idle source reports, so it cannot be used to tell the two apart.
## Evidence
Reproduced deterministically on Pulsar 3.3.7 against a Debezium-based source pointed at an
unreachable database, sampling `sources status` every 10 s for 60 s. Two failure shapes were
tested and behave identically:
- **blackholed** — `192.0.2.10` (RFC 5737 TEST-NET-1), packets discarded, connects hang
- **refused** — `127.0.0.1:7100`, immediate `ECONNREFUSED`
Both held `running: true` / `error: ""` / `numRestarts: 0` / `numReceivedFromSource: 0` for
the entire window. The instance log over the same period contained 37 INFO, 10 WARN and
**zero ERROR** lines. Startup ran cleanly all the way through `KafkaConnectSource`,
`PulsarKafkaWorkerConfig` and `PulsarOffsetBackingStore` before stalling, so every signal
the adaptor emits says the source is fine.
## Scope
This is not specific to one connector. Any source using the adaptor inherits it, including
the Debezium MySQL/Postgres connectors in this repo — a Debezium source with a wrong
`database.hostname` should reproduce it.
To be explicit about what is *not* being reported here: the underlying client retrying
forever without escalating is the connector's own behaviour, not the adaptor's, and stock
Kafka Connect would also report such a task as RUNNING. The gap being raised is that the
adaptor holds the information needed to distinguish "has never produced anything" from
"idle" and currently throws it away.
## Suggested direction
Something cheap and non-behavioural would be enough:
- track consecutive empty polls and the time of the last non-empty poll in
`AbstractKafkaConnectSource`
- expose them via the existing source metrics, so `sources status` can distinguish
"never produced a record since start" from "idle"
- optionally log at a decaying interval once the empty-poll streak crosses a threshold
Happy to put up a PR if the direction seems right.
## Environment
- Pulsar 3.3.7 (`kafka-connect-adaptor`); the `read()` loop above is unchanged on master
- JDK 17, source deployed as a NAR to a Pulsar worker
Contributor guide
No contributing guide indexed for this repository
Research direction
Start in AbstractKafkaConnectSource.read() and trace the existing source metrics surfaced by pulsar-admin sources status. Reproduce with the unreachable or refused Debezium database examples, then verify that empty-poll and last-non-empty state distinguish a never-producing source from an idle one without changing connector retry behavior.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java, kafka
- Domain
- distributed-systems, stream-processing
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 48/100