This is an automated email from the ASF dual-hosted git repository.
david-streamlio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-connectors.git
The following commit(s) were added to refs/heads/master by this push:
new 661efb55 [fix][io] Flush and sync before closing the HDFS text sink
stream (#100)
661efb55 is described below
commit 661efb55d344360314d02da82a7e789fab44e763
Author: David Kjerrumgaard <[email protected]>
AuthorDate: Fri Jul 10 14:54:45 2026 -0700
[fix][io] Flush and sync before closing the HDFS text sink stream (#100)
* [fix][io] Flush and sync before closing the HDFS text sink stream
HdfsAbstractTextFileSink.close() closed the writer (and its underlying
HDFS stream) before the superclass ran its final hsync()/ack in
HdfsSyncThread.halt(). Whenever records were still unacked at close
time, that hsync() ran against an already-closed stream and threw
ClosedChannelException, so the final batch was never acked or committed.
Flush the writer, let super.close() halt the sync thread and hsync/ack
while the stream is still open, then close the writer.
Add a regression test that closes the sink with records still queued
(a long syncInterval keeps the background thread from draining first):
it fails with ClosedChannelException against the old order and passes
with the fix.
* [fix][test] Address review on the HDFS close-ordering fix
- close() now wraps flush()+super.close() in try/finally so the writer
(and its underlying HDFS stream) is always closed, even if the final
hsync()/ack in super.close() throws.
- The regression test mirrors the defensive close-on-error pattern so a
throwing close() cannot leave the non-daemon HdfsSyncThread running.
- Lower the test's syncInterval from 5s to 1s: close() joins the sync
thread (halt() does not interrupt its sleep), so the interval is dead
suite time; 1s stays comfortably above the millisecond enqueue.
---
.../hdfs3/sink/text/HdfsAbstractTextFileSink.java | 17 ++++-
.../io/hdfs3/sink/HdfsSinkIntegrationTest.java | 75 ++++++++++++++++++++--
2 files changed, 84 insertions(+), 8 deletions(-)
diff --git
a/hdfs3/src/main/java/org/apache/pulsar/io/hdfs3/sink/text/HdfsAbstractTextFileSink.java
b/hdfs3/src/main/java/org/apache/pulsar/io/hdfs3/sink/text/HdfsAbstractTextFileSink.java
index 66fd7a62..1487bdd8 100644
---
a/hdfs3/src/main/java/org/apache/pulsar/io/hdfs3/sink/text/HdfsAbstractTextFileSink.java
+++
b/hdfs3/src/main/java/org/apache/pulsar/io/hdfs3/sink/text/HdfsAbstractTextFileSink.java
@@ -48,8 +48,21 @@ public abstract class HdfsAbstractTextFileSink<K, V> extends
HdfsAbstractSink<K,
@Override
public void close() throws Exception {
- writer.close();
- super.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. Closing
the writer first (the
+ // previous order) closed the stream before that final hsync(),
throwing
+ // ClosedChannelException whenever unacked records remained at close
time. The writer is
+ // closed in finally so the underlying stream is released even if
super.close() throws.
+ try {
+ if (writer != null) {
+ writer.flush();
+ }
+ super.close();
+ } finally {
+ if (writer != null) {
+ writer.close();
+ }
+ }
}
@Override
diff --git
a/hdfs3/src/test/java/org/apache/pulsar/io/hdfs3/sink/HdfsSinkIntegrationTest.java
b/hdfs3/src/test/java/org/apache/pulsar/io/hdfs3/sink/HdfsSinkIntegrationTest.java
index 4106a6e3..e7942bd0 100644
---
a/hdfs3/src/test/java/org/apache/pulsar/io/hdfs3/sink/HdfsSinkIntegrationTest.java
+++
b/hdfs3/src/test/java/org/apache/pulsar/io/hdfs3/sink/HdfsSinkIntegrationTest.java
@@ -155,7 +155,7 @@ public class HdfsSinkIntegrationTest {
}
// Read the committed file back from the mini cluster's filesystem
and assert it matches.
- String actual = readSinkOutput();
+ String actual = readSinkOutput(DIRECTORY);
Assert.assertEquals(actual, expected.toString(),
"content read back from HDFS must match what was written
to the sink");
} finally {
@@ -171,12 +171,75 @@ public class HdfsSinkIntegrationTest {
}
/**
- * 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()
+ // (enqueuing the records takes milliseconds), so close() runs the
final hsync()/ack path
+ // with records still queued — the bug scenario. Kept small because
close() joins the sync
+ // thread, which sleeps this long before exiting, so it directly adds
to the suite runtime.
+ config.put("syncInterval", 1_000L);
+
+ SinkContext sinkContext = mock(SinkContext.class);
+ AtomicInteger ackCount = new AtomicInteger(0);
+
+ HdfsStringSink sink = new HdfsStringSink();
+ sink.open(config, sinkContext);
+ boolean closed = false;
+ try {
+ for (String value : values) {
+ sink.write(mockRecord(value, ackCount));
+ }
+
+ // Under the old ordering this threw ClosedChannelException; it
must now commit cleanly.
+ sink.close();
+ closed = true;
+
+ 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");
+ } finally {
+ if (!closed) {
+ // Failure path: close so the non-daemon HdfsSyncThread cannot
outlive the test.
+ try {
+ sink.close();
+ } catch (Exception e) {
+ log.warn("Failed to close sink on the error path", e);
+ }
+ }
+ }
+ }
+
+ /**
+ * Reads and concatenates the content of every file the sink produced
under {@code directory}
+ * on the mini cluster's filesystem, matching files whose name starts with
+ * {@link #FILENAME_PREFIX}.
+ */
+ private String readSinkOutput(String directory) throws Exception {
+ Path dir = new Path(directory);
if (!clusterFs.exists(dir)) {
return "";
}
