You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "david-streamlio (via GitHub)" <gi...@apache.org> on 2026/07/10 19:45:42 UTC
[PR] [fix][io] Flush and sync before closing the HDFS text sink stream [pulsar-connectors]
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Flush and sync before closing the HDFS text sink stream [pulsar-connectors]
Posted by "david-streamlio (via GitHub)" <gi...@apache.org>.
david-streamlio merged PR #100:
URL: https://github.com/apache/pulsar-connectors/pull/100
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] Flush and sync before closing the HDFS text sink stream [pulsar-connectors]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #100:
URL: https://github.com/apache/pulsar-connectors/pull/100#discussion_r3561616266
##########
hdfs3/src/test/java/org/apache/pulsar/io/hdfs3/sink/HdfsSinkIntegrationTest.java:
##########
@@ -171,12 +171,62 @@ public void testRecordsAreWrittenToHdfs() throws Exception {
}
/**
- * Reads and concatenates the content of every file produced by the sink under {@link #DIRECTORY}
- * on the mini cluster's filesystem. The sink writes all records from a single {@code open()} into
- * one file whose name starts with {@link #FILENAME_PREFIX}.
+ * Regression test for the close ordering bug: closing the sink while records are still unacked
+ * must flush and commit them, not throw. The old {@code writer.close()} before the superclass'
+ * final {@code hsync()} closed the stream first, so {@code hsync()} threw {@code
+ * ClosedChannelException} whenever the sync thread had not yet drained the queue at close time.
*/
- private String readSinkOutput() throws Exception {
- Path dir = new Path(DIRECTORY);
+ @Test(timeOut = 300_000)
+ public void testCloseWithUnackedRecordsCommitsInsteadOfThrowing() throws Exception {
+ String directory = "/hdfs-sink-close-ordering-test";
+ List<String> values = new ArrayList<>();
+ for (int i = 0; i < 25; i++) {
+ values.add("close-record-" + i);
+ }
+
+ Map<String, Object> config = new HashMap<>();
+ config.put("hdfsConfigResources", coreSite.getAbsolutePath());
+ config.put("directory", directory);
+ config.put("filenamePrefix", FILENAME_PREFIX);
+ config.put("fileExtension", FILE_EXTENSION);
+ config.put("separator", SEPARATOR);
+ config.put("encoding", "UTF-8");
+ // A sync interval long enough that the background thread does not tick before we close()
+ // (writes take milliseconds), so close() runs the final hsync()/ack path with records still
+ // queued — the bug scenario. Kept modest because close() joins the sync thread, which sleeps
+ // this long before exiting.
+ config.put("syncInterval", 5_000L);
+
+ SinkContext sinkContext = mock(SinkContext.class);
+ AtomicInteger ackCount = new AtomicInteger(0);
+
+ HdfsStringSink sink = new HdfsStringSink();
+ sink.open(config, sinkContext);
+ for (String value : values) {
+ sink.write(mockRecord(value, ackCount));
+ }
+
+ // Under the old ordering this threw ClosedChannelException; it must now commit cleanly.
+ sink.close();
+
+ Assert.assertEquals(ackCount.get(), values.size(),
+ "close() should have flushed and acked every queued record");
+
+ StringBuilder expected = new StringBuilder();
+ for (String value : values) {
+ expected.append(value).append(SEPARATOR);
+ }
+ Assert.assertEquals(readSinkOutput(directory), expected.toString(),
+ "records written before close() must be committed to HDFS");
+ }
Review Comment:
The new regression test calls sink.close() without a failure-path cleanup. If sink.close() throws (the behavior this test is guarding against), the non-daemon HdfsSyncThread may keep running and can delay or hang the test JVM. Mirror the defensive close-on-error pattern used in testRecordsAreWrittenToHdfs().
##########
hdfs3/src/main/java/org/apache/pulsar/io/hdfs3/sink/text/HdfsAbstractTextFileSink.java:
##########
@@ -48,8 +48,17 @@ protected void createWriter() throws IOException {
@Override
public void close() throws Exception {
- writer.close();
+ // Flush buffered bytes to the HDFS stream, then let the superclass halt the sync thread and
+ // run its final hsync()/ack while the stream is still open, and only then close the writer.
+ // Closing the writer first (the previous order) closed the stream before that final
+ // hsync(), throwing ClosedChannelException whenever unacked records remained at close time.
+ if (writer != null) {
+ writer.flush();
+ }
super.close();
+ if (writer != null) {
+ writer.close();
+ }
Review Comment:
close() now closes the writer only after super.close(). If super.close() throws (e.g., an IOException from the final hsync/ack path), writer.close() will be skipped, leaking the underlying HDFS stream. Consider wrapping the shutdown sequence in try/finally so the writer is always closed, and preserve/suppress exceptions appropriately.
##########
hdfs3/src/test/java/org/apache/pulsar/io/hdfs3/sink/HdfsSinkIntegrationTest.java:
##########
@@ -171,12 +171,62 @@ public void testRecordsAreWrittenToHdfs() throws Exception {
}
/**
- * Reads and concatenates the content of every file produced by the sink under {@link #DIRECTORY}
- * on the mini cluster's filesystem. The sink writes all records from a single {@code open()} into
- * one file whose name starts with {@link #FILENAME_PREFIX}.
+ * Regression test for the close ordering bug: closing the sink while records are still unacked
+ * must flush and commit them, not throw. The old {@code writer.close()} before the superclass'
+ * final {@code hsync()} closed the stream first, so {@code hsync()} threw {@code
+ * ClosedChannelException} whenever the sync thread had not yet drained the queue at close time.
*/
- private String readSinkOutput() throws Exception {
- Path dir = new Path(DIRECTORY);
+ @Test(timeOut = 300_000)
+ public void testCloseWithUnackedRecordsCommitsInsteadOfThrowing() throws Exception {
+ String directory = "/hdfs-sink-close-ordering-test";
+ List<String> values = new ArrayList<>();
+ for (int i = 0; i < 25; i++) {
+ values.add("close-record-" + i);
+ }
+
+ Map<String, Object> config = new HashMap<>();
+ config.put("hdfsConfigResources", coreSite.getAbsolutePath());
+ config.put("directory", directory);
+ config.put("filenamePrefix", FILENAME_PREFIX);
+ config.put("fileExtension", FILE_EXTENSION);
+ config.put("separator", SEPARATOR);
+ config.put("encoding", "UTF-8");
+ // A sync interval long enough that the background thread does not tick before we close()
+ // (writes take milliseconds), so close() runs the final hsync()/ack path with records still
+ // queued — the bug scenario. Kept modest because close() joins the sync thread, which sleeps
+ // this long before exiting.
+ config.put("syncInterval", 5_000L);
Review Comment:
syncInterval is set to 5 seconds to keep the background thread from draining the queue before close(), but in this implementation close() waits for HdfsSyncThread to exit (halt() does not interrupt sleep), so this test will typically block for ~syncInterval and add ~5s to the suite. Consider lowering the interval to reduce test runtime while still being comfortably above the time it takes to enqueue the records.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org