danny0405 commented on code in PR #19540:
URL: https://github.com/apache/hudi/pull/19540#discussion_r3734789016
##########
hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/utils/TestStreamerUtil.java:
##########
@@ -273,4 +294,263 @@ void testTableExist() throws IOException {
assertTrue(StreamerUtil.tableExists(basePath,
HadoopConfigurations.getHadoopConf(conf)));
}
}
+
+ @Test
+ void testBuildProperties() {
+ TypedProperties properties = StreamerUtil.buildProperties(
+ Arrays.asList("hoodie.test.one=1", "hoodie.test.two=two"));
+
+ assertEquals("1", properties.getString("hoodie.test.one"));
+ assertEquals("two", properties.getString("hoodie.test.two"));
+ assertThrows(IllegalArgumentException.class,
+ () ->
StreamerUtil.buildProperties(Collections.singletonList("invalid")));
+ }
+
+ @Test
+ void testSourceSchemaConfiguration() {
+ Configuration conf = new Configuration();
+ String schema =
"{\"type\":\"record\",\"name\":\"record\",\"fields\":[{\"name\":\"id\",\"type\":\"long\"}]}";
+ conf.set(FlinkOptions.SOURCE_AVRO_SCHEMA, schema);
+
+ HoodieSchema sourceSchema = StreamerUtil.getSourceSchema(conf);
+
+ assertTrue(sourceSchema.getField("id").isPresent());
+ assertThrows(HoodieException.class,
+ () -> StreamerUtil.getSourceSchema(new Configuration()));
+ }
+
+ @Test
+ void testConfigurationConversions() {
+ Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ conf.set(FlinkOptions.COMPACTION_MAX_MEMORY, 256);
+ conf.set(FlinkOptions.ORDERING_FIELDS, "ts");
+ conf.set(FlinkOptions.INDEX_TYPE, HoodieIndex.IndexType.BLOOM.name());
+
+ TypedProperties properties = StreamerUtil.flinkConf2TypedProperties(conf);
+
+ assertEquals(conf.get(FlinkOptions.TABLE_TYPE),
+ properties.getString(HoodieTableConfig.TYPE.key()));
+ assertEquals(256L * 1024 * 1024,
StreamerUtil.getMaxCompactionMemoryInBytes(conf));
+ assertEquals("ts",
StreamerUtil.getPayloadConfig(conf).getString(HoodiePayloadConfig.ORDERING_FIELDS));
+ assertEquals(HoodieIndex.IndexType.BLOOM.name(),
+
StreamerUtil.getIndexConfig(conf).getString(HoodieIndexConfig.INDEX_TYPE));
+ }
+
+ @Test
+ void testSimplePathAndFileUtilities() {
+ assertEquals("partition_file-id",
StreamerUtil.generateBucketKey("partition", "file-id"));
+
+ assertFalse(StreamerUtil.isValidFile(pathInfo("file.parquet", 4)));
+ assertTrue(StreamerUtil.isValidFile(pathInfo("file.parquet", 5)));
+ assertFalse(StreamerUtil.isValidFile(pathInfo("file.log", 6)));
+ assertTrue(StreamerUtil.isValidFile(pathInfo("file.log", 7)));
+ assertFalse(StreamerUtil.isValidFile(pathInfo("file.orc", 3)));
+ assertTrue(StreamerUtil.isValidFile(pathInfo("file.orc", 4)));
+ assertFalse(StreamerUtil.isValidFile(pathInfo("file.unknown", 0)));
+ assertTrue(StreamerUtil.isValidFile(pathInfo("file.unknown", 1)));
+ }
+
+ @Test
+ void testPartitionExists() throws IOException {
+ Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ org.apache.hadoop.conf.Configuration hadoopConf =
HadoopConfigurations.getHadoopConf(conf);
+ assertFalse(StreamerUtil.partitionExists(tempFile.getAbsolutePath(),
"dt=2026-08-06", hadoopConf));
+
+ try (FileSystem fs = HadoopFSUtils.getFs(tempFile.getAbsolutePath(),
hadoopConf)) {
Review Comment:
`HadoopFSUtils.getFs` ultimately calls `Path#getFileSystem`, which may
return a cached/shared `FileSystem`. Closing that instance here can invalidate
other users of the same cached filesystem in this JVM and make the new test
interfere with parallel tests. Could this use an owned
`FileSystem.newInstance(...)`/`newInstanceLocal(...)` handle, or avoid closing
the shared handle?
--
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]
