Copilot commented on code in PR #93: URL: https://github.com/apache/pulsar-connectors/pull/93#discussion_r3560742257
########## hdfs3/src/test/java/org/apache/pulsar/io/hdfs3/sink/HdfsSinkIntegrationTest.java: ########## @@ -0,0 +1,195 @@ +/* + * 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. + */ +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); + // Let the background sync thread flush and ack on a modest interval. + config.put("syncInterval", 200L); + Review Comment: The test decodes the written HDFS bytes as UTF-8, but the sink will encode records using `Charset.defaultCharset()` unless the `encoding` config is set (see AbstractHdfsConnector#getEncoding). On environments where the default charset is not UTF-8, this can make the test fail (or worse, make it pass locally but fail in CI). Set `encoding` explicitly to UTF-8 in the sink config so the write and read sides use the same charset deterministically. ########## hdfs3/build.gradle.kts: ########## @@ -21,6 +21,16 @@ 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 affects the test classpath only; the connector itself is an HDFS +// client and never loads Jetty at runtime. Review Comment: This comment says excluding the Jetty 12 BOM “affects the test classpath only”, but `pulsarConnectorsDependencies { exclude(...) }` customizes the enforced platform attached to the module’s `implementation` configuration (and therefore influences main compile/runtime resolution too, even if Jetty isn’t ultimately present). Adjust the wording to avoid implying the exclusion is test-scoped. ########## hdfs3/src/test/java/org/apache/pulsar/io/hdfs3/sink/HdfsSinkIntegrationTest.java: ########## @@ -0,0 +1,195 @@ +/* + * 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. + */ +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); + // 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); + 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(); + Review Comment: If the Awaitility assertion times out (or any earlier assertion throws), `sink.close()` won’t run, leaving `HdfsSyncThread` (a non-daemon thread) running and potentially preventing the test JVM from terminating cleanly. Wrap the sink lifecycle in a try/finally and only close when `open()` succeeds so resources/threads are always cleaned up on failure paths too. -- 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]
