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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
