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 d32fb93b [improve][test] Add HDFS3 sink integration test against
MiniDFSCluster (#93)
d32fb93b is described below
commit d32fb93b643b9c9673caa105a25169d30c1be62c
Author: David Kjerrumgaard <[email protected]>
AuthorDate: Fri Jul 10 12:26:51 2026 -0700
[improve][test] Add HDFS3 sink integration test against MiniDFSCluster (#93)
* [improve][test] Add HDFS3 sink integration test against MiniDFSCluster
Adds an end-to-end integration test for the HDFS3 sink that exercises the
sink against a real, in-JVM Hadoop MiniDFSCluster (no Docker required).
Previously the hdfs3 module had only config tests and mock-based sink
tests that verified acks against a Mockito mock and never wrote to a real
filesystem.
The test opens the HdfsStringSink pointed at the mini cluster via a
generated core-site.xml, writes several records, waits for them to be
acked, closes the sink to flush, then reads the committed file back from
the mini cluster's own FileSystem and asserts the bytes match what was
written.
Excludes the Jetty 12 BOM from the enforced platform for this module and
pins the test classpath to Jetty 9.x (via the existing jetty9 override)
so the mini cluster's embedded HDFS HTTP server can start.
* [fix][test] Address review on HdfsSinkIntegrationTest
- Set the sink's encoding to UTF-8 explicitly so the write side matches
the UTF-8 read-back regardless of the JVM default charset.
- Wrap the sink lifecycle in try/finally so close() runs on failure
paths too, preventing a leaked non-daemon HdfsSyncThread.
- Correct the build.gradle.kts comment: the Jetty BOM exclusion
customizes the enforced platform on the implementation configuration,
not just the test classpath (no Jetty reaches the runtime either way).
---
hdfs3/build.gradle.kts | 16 ++
.../io/hdfs3/sink/HdfsSinkIntegrationTest.java | 214 +++++++++++++++++++++
2 files changed, 230 insertions(+)
diff --git a/hdfs3/build.gradle.kts b/hdfs3/build.gradle.kts
index 82e22541..0f23ad14 100644
--- a/hdfs3/build.gradle.kts
+++ b/hdfs3/build.gradle.kts
@@ -21,6 +21,18 @@ plugins {
id("pulsar-connectors.java-conventions")
id("pulsar-connectors.nar-conventions")
}
+
+// The hadoop-minicluster used by the sink integration test embeds an HDFS
namenode whose HTTP
+// server requires Jetty 9.x, while the shared platform enforces Jetty 12.x
(which removed classes
+// such as HandlerWrapper). Drop the Jetty 12 BOM from the enforced platform
so the Jetty 9.x
+// override below can apply. This customizes the enforced platform on the
module's implementation
+// configuration, so it affects main compile/runtime resolution as well as
tests — but the
+// connector is an HDFS client that never loads Jetty, so no Jetty ends up on
the runtime classpath
+// either way; the change only unblocks the mini cluster's Jetty 9.x on the
test classpath.
+pulsarConnectorsDependencies {
+ exclude(libs.jetty.bom)
+}
+
dependencies {
implementation(libs.pulsar.io.core)
implementation(libs.jackson.databind)
@@ -32,4 +44,8 @@ dependencies {
}
implementation(libs.bcprov.jdk18on)
implementation(libs.jakarta.activation.api)
+
+ // In-JVM HDFS cluster for sink integration testing (no external services
/ Docker required).
+ testImplementation(enforcedPlatform(libs.jetty9.bom.override))
+ testImplementation(libs.hadoop.minicluster)
}
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
new file mode 100644
index 00000000..4106a6e3
--- /dev/null
+++
b/hdfs3/src/test/java/org/apache/pulsar/io/hdfs3/sink/HdfsSinkIntegrationTest.java
@@ -0,0 +1,214 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.hdfs3.sink;
+
+import static org.awaitility.Awaitility.await;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.io.core.SinkContext;
+import org.apache.pulsar.io.hdfs3.sink.text.HdfsStringSink;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+/**
+ * Integration test for the HDFS sink, exercised against a real, in-JVM Hadoop
+ * {@link MiniDFSCluster}.
+ *
+ * <p>Unlike the existing sink unit tests (which point at the local filesystem
and only verify that
+ * records are acknowledged against a Mockito mock), this test drives the
{@link HdfsStringSink}
+ * end-to-end against an actual HDFS namenode/datanode: it opens the sink,
writes several records,
+ * closes the sink to flush, then reads the produced file back <em>from the
mini cluster's own
+ * {@link FileSystem}</em> and asserts the bytes on HDFS match exactly what
was written.
+ *
+ * <p>No Docker container is required — {@code MiniDFSCluster} runs entirely
inside the test JVM.
+ */
[email protected]
+public class HdfsSinkIntegrationTest {
+
+ private static final String DIRECTORY = "/hdfs-sink-integration-test";
+ private static final String FILENAME_PREFIX = "records";
+ private static final String FILE_EXTENSION = ".txt";
+ private static final char SEPARATOR = '\n';
+
+ private MiniDFSCluster cluster;
+ private FileSystem clusterFs;
+ private File baseDir;
+ private File coreSite;
+
+ @BeforeClass(alwaysRun = true)
+ public void setUp() throws Exception {
+ baseDir = Files.createTempDirectory("hdfs-sink-it").toFile();
+
+ Configuration conf = new Configuration();
+ conf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR,
baseDir.getAbsolutePath());
+ cluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).build();
+ cluster.waitActive();
+ clusterFs = cluster.getFileSystem();
+
+ // The sink reads its Hadoop configuration from files named in
`hdfsConfigResources`. Write a
+ // core-site.xml whose fs.defaultFS points at the running mini cluster
so the sink connects to
+ // it rather than to the local filesystem.
+ String defaultFs = clusterFs.getUri().toString();
+ coreSite = new File(baseDir, "core-site.xml");
+ String xml = "<?xml version=\"1.0\"?>\n"
+ + "<configuration>\n"
+ + " <property>\n"
+ + " <name>fs.defaultFS</name>\n"
+ + " <value>" + defaultFs + "</value>\n"
+ + " </property>\n"
+ + "</configuration>\n";
+ Files.write(coreSite.toPath(), xml.getBytes(StandardCharsets.UTF_8));
+ }
+
+ @AfterClass(alwaysRun = true)
+ public void tearDown() throws Exception {
+ if (cluster != null) {
+ cluster.shutdown(true);
+ }
+ }
+
+ @Test(timeOut = 300_000)
+ public void testRecordsAreWrittenToHdfs() throws Exception {
+ List<String> values = new ArrayList<>();
+ for (int i = 0; i < 25; i++) {
+ values.add("integration-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);
+ // Encode explicitly as UTF-8 so the write side matches the UTF-8
decode in readSinkOutput()
+ // regardless of the JVM's default charset (the sink otherwise uses
Charset.defaultCharset()).
+ config.put("encoding", "UTF-8");
+ // Let the background sync thread flush and ack on a modest interval.
+ config.put("syncInterval", 200L);
+
+ SinkContext sinkContext = mock(SinkContext.class);
+ AtomicInteger ackCount = new AtomicInteger(0);
+
+ HdfsStringSink sink = new HdfsStringSink();
+ sink.open(config, sinkContext);
+ // close() must run even if an assertion below fails, otherwise the
non-daemon
+ // HdfsSyncThread keeps running and can stop the test JVM from
terminating cleanly.
+ boolean closed = false;
+ try {
+ for (String value : values) {
+ sink.write(mockRecord(value, ackCount));
+ }
+
+ // Wait until every record has been acked. The sink acks a record
only after its
+ // background sync thread has hsync'd the stream, which also
guarantees the sink's
+ // unacked-record queue has drained — so the subsequent close()
flushes the writer and
+ // commits the file cleanly instead of racing an hsync against an
already-closed stream.
+ await().atMost(60, TimeUnit.SECONDS).pollInterval(200,
TimeUnit.MILLISECONDS)
+ .untilAsserted(() -> Assert.assertEquals(ackCount.get(),
values.size(),
+ "all records should be acknowledged by the sink"));
+
+ // close() flushes the buffered writer and commits the file to
HDFS.
+ sink.close();
+ closed = true;
+
+ // Build the content we expect on HDFS: each value followed by the
separator.
+ StringBuilder expected = new StringBuilder();
+ for (String value : values) {
+ expected.append(value).append(SEPARATOR);
+ }
+
+ // Read the committed file back from the mini cluster's filesystem
and assert it matches.
+ String actual = readSinkOutput();
+ Assert.assertEquals(actual, expected.toString(),
+ "content read back from HDFS must match what was written
to the sink");
+ } finally {
+ if (!closed) {
+ // Best-effort cleanup on a failure path; the real assertion
has already thrown.
+ 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 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}.
+ */
+ private String readSinkOutput() throws Exception {
+ Path dir = new Path(DIRECTORY);
+ if (!clusterFs.exists(dir)) {
+ return "";
+ }
+ StringBuilder content = new StringBuilder();
+ for (FileStatus status : clusterFs.listStatus(dir)) {
+ String name = status.getPath().getName();
+ if (!status.isFile() || !name.startsWith(FILENAME_PREFIX)) {
+ continue;
+ }
+ try (FSDataInputStream in = clusterFs.open(status.getPath())) {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ byte[] buffer = new byte[4096];
+ int read;
+ while ((read = in.read(buffer)) != -1) {
+ out.write(buffer, 0, read);
+ }
+ content.append(new String(out.toByteArray(),
StandardCharsets.UTF_8));
+ }
+ }
+ return content.toString();
+ }
+
+ @SuppressWarnings("unchecked")
+ private Record<String> mockRecord(String value, AtomicInteger ackCount) {
+ Record<String> record = mock(Record.class);
+ when(record.getValue()).thenReturn(value);
+ when(record.getKey()).thenReturn(Optional.of(value));
+ // Count acks so the test can wait for the sink to fully drain before
closing.
+ doAnswer(invocation -> {
+ ackCount.incrementAndGet();
+ return null;
+ }).when(record).ack();
+ return record;
+ }
+}
