david-streamlio opened a new pull request, #100:
URL: https://github.com/apache/pulsar-connectors/pull/100
Fixes #94
### Motivation
`HdfsAbstractTextFileSink.close()` closed the writer — and with it the
underlying HDFS `FSDataOutputStream` — **before** the superclass ran its final
flush:
```java
public void close() throws Exception {
writer.close(); // closes the stream
super.close(); // -> HdfsAbstractSink.close() -> syncThread.halt()
-> ackRecords() -> stream.hsync()
}
```
`HdfsSyncThread.halt()` runs a final `ackRecords()`, which does
`stream.hsync()` when any records are still unacked. Because `writer.close()`
already closed that stream, the `hsync()` threw `ClosedChannelException` — so
on shutdown with a non-empty unacked queue, the final batch was neither synced
nor acked.
It only reproduces against real HDFS, which is why the existing mock-based
sink tests (local filesystem, which does not throw on hsync-after-close) never
caught it. Found while adding the HDFS integration test in #93, whose test
sidesteps it by draining the queue before close.
### Modifications
Reorder `close()` so the final `hsync()`/ack happens while the stream is
still open:
```java
if (writer != null) {
writer.flush(); // push buffered bytes to the stream
}
super.close(); // halt sync thread; final hsync()/ack on the OPEN
stream
if (writer != null) {
writer.close(); // now close the stream
}
```
### Verifying this change
Added `testCloseWithUnackedRecordsCommitsInsteadOfThrowing` to
`HdfsSinkIntegrationTest`: it writes records and closes immediately, with a
long `syncInterval` so the background thread has not drained the queue — the
exact bug scenario. Verified in both directions against the real MiniDFSCluster:
- **Old order** → `testCloseWithUnackedRecordsCommitsInsteadOfThrowing
FAILED — java.nio.channels.ClosedChannelException`
- **Fixed order** → both integration tests `PASSED`, `BUILD SUCCESSFUL`
So the test genuinely guards the fix rather than passing alongside it.
--
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]
