david-streamlio opened a new issue, #94:
URL: https://github.com/apache/pulsar-connectors/issues/94
## Description
The HDFS3 text sink can throw `ClosedChannelException` during `close()` when
records are still unacked. Found while adding integration coverage for the
module (#52 / #93) — it does not reproduce against the existing mock tests
because they write to the local filesystem, which does not throw on `hsync()`
after close.
## Root cause
`HdfsAbstractTextFileSink.close()` closes the output stream first, then
delegates to the superclass, which flushes it:
```java
// HdfsAbstractTextFileSink.close()
public void close() throws Exception {
writer.close(); // closes the underlying FSDataOutputStream
super.close(); // -> HdfsAbstractSink.close() -> syncThread.halt()
}
```
`syncThread.halt()` runs a final `ackRecords()`, which calls
`stream.hsync()` on any still-unacked batch:
```java
// HdfsSyncThread.ackRecords()
if (stream != null) {
stream.hsync(); // <-- stream may already be closed by writer.close()
}
while (!unackedRecords.isEmpty()) {
unackedRecords.take().ack();
}
```
The guard checks only that the stream is non-null, not that it is still
open. So when `close()` is called with records the sync thread has not yet
flushed, `writer.close()` closes the stream and the subsequent `hsync()` throws
`ClosedChannelException` against a real HDFS `FSDataOutputStream`.
## Impact
On shutdown with a non-empty unacked queue, the final flush fails and those
records may not be acked/committed. It surfaces only against real HDFS, which
is why the mock-based tests never caught it.
## Suggested fix
Order shutdown so the final sync/ack happens before the stream is closed —
e.g. `super.close()` (halt the sync thread, flush, ack) *then* `writer.close()`
— or make `ackRecords()` skip `hsync()` when the stream is already closed.
## Workaround in #93
The integration test in #93 sidesteps this by waiting for the unacked queue
to fully drain before calling `close()`, so the test is deterministic. The
underlying ordering bug is left for this issue rather than fixed there (out of
scope for a test-only PR).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
