You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by yi...@apache.org on 2024/05/15 12:21:12 UTC
(hudi) branch branch-0.x updated: [HUDI-7587] Make hudi-hadoop-common module dependent on hudi-common module (#11131)
This is an automated email from the ASF dual-hosted git repository.
yihua pushed a commit to branch branch-0.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/branch-0.x by this push:
new c56fed9620d [HUDI-7587] Make hudi-hadoop-common module dependent on hudi-common module (#11131)
c56fed9620d is described below
commit c56fed9620d8ae5b8a7914dab6653a891fdf6179
Author: Jon Vexler <jb...@gmail.com>
AuthorDate: Wed May 15 05:13:27 2024 -0700
[HUDI-7587] Make hudi-hadoop-common module dependent on hudi-common module (#11131)
Co-authored-by: Jonathan Vexler <=>
---
hudi-cli/pom.xml | 7 +
hudi-client/hudi-client-common/pom.xml | 25 +
.../java/org/apache/hudi/table/HoodieTable.java | 4 +-
.../transaction/TestInProcessLockProvider.java | 527 ---------------------
hudi-client/hudi-flink-client/pom.xml | 13 +
hudi-client/hudi-java-client/pom.xml | 13 +
hudi-client/hudi-spark-client/pom.xml | 28 ++
.../java/org/apache/hudi/client/TestMultiFS.java | 3 +-
.../apache/hudi/table/TestConsistencyGuard.java | 2 +-
.../hudi/table/marker/TestDirectWriteMarkers.java | 4 +-
.../hudi/table/marker/TestWriteMarkersBase.java | 6 +-
hudi-common/pom.xml | 6 -
.../bootstrap/index/HFileBootstrapIndex.java | 2 +-
.../apache/hudi/common/config/HoodieConfig.java | 3 +-
.../hudi/common/config/PropertiesConfig.java | 33 +-
.../java/org/apache/hudi/common/fs/FSUtils.java | 68 +--
.../hudi/common/fs/FailSafeConsistencyGuard.java | 1 -
.../hudi/common}/fs/NoOpConsistencyGuard.java | 2 +-
.../hudi/common/fs/OptimisticConsistencyGuard.java | 1 -
.../hudi/common/table/HoodieTableMetaClient.java | 7 +-
.../hudi/common/table/log/HoodieLogFileReader.java | 2 +-
.../table/log/block/HoodieHFileDataBlock.java | 10 +-
.../table/log/block/HoodieParquetDataBlock.java | 11 +-
.../org/apache/hudi/common/util/BaseFileUtils.java | 5 +-
.../org/apache/hudi/common/util/ConfigUtils.java | 5 +
.../org/apache/hudi/common/util/ParquetUtils.java | 9 +-
.../io/storage/HoodieAvroFileWriterFactory.java | 31 +-
.../io/storage/HoodieNativeAvroHFileReader.java | 4 +-
.../main/java/org/apache/hudi/metrics/Metrics.java | 4 +-
.../apache/hudi/storage/HoodieStorageUtils.java | 48 +-
.../apache/hudi/common/fs/TestFSUtilsMocked.java | 120 -----
.../common/testutils/HoodieTestDataGenerator.java | 24 +-
.../hudi/common/testutils/HoodieTestUtils.java | 14 +-
hudi-examples/hudi-examples-common/pom.xml | 5 +
hudi-examples/hudi-examples-flink/pom.xml | 13 +
hudi-flink-datasource/hudi-flink/pom.xml | 13 +
hudi-hadoop-common/pom.xml | 30 ++
.../common/config/DFSPropertiesConfiguration.java | 7 +-
.../java/org/apache/hudi/common/util/OrcUtils.java | 15 +-
.../org/apache/hudi/hadoop/fs/HadoopFSUtils.java | 41 +-
.../hudi/hadoop/fs/HoodieWrapperFileSystem.java | 11 +-
.../hadoop/fs/SizeAwareFSDataOutputStream.java | 6 +-
.../hudi/hadoop/fs/inline/HadoopInLineFSUtils.java | 66 +++
.../hudi/hadoop/fs/inline/InLineFileSystem.java | 15 +-
.../hudi/io/storage/HoodieAvroHFileWriter.java | 4 +-
.../hudi/io/storage/HoodieAvroOrcWriter.java | 8 +-
.../hudi/io/storage/HoodieAvroParquetWriter.java | 0
.../hudi/io/storage/HoodieBaseParquetWriter.java | 4 +-
.../storage/hadoop/HadoopStorageConfiguration.java | 10 +-
.../hudi/storage/hadoop/HoodieHadoopStorage.java | 49 +-
.../apache/hudi/common/bloom/TestBloomFilter.java | 2 +-
.../hudi/common/bootstrap/TestBootstrapIndex.java | 0
.../org/apache/hudi/common/fs/TestFSUtils.java | 15 +-
.../fs/TestFSUtilsWithRetryWrapperEnable.java | 1 -
.../common/fs/TestHoodieWrapperFileSystem.java | 1 -
.../apache/hudi/common/fs/TestStorageSchemes.java | 0
.../hudi/common/fs/inline/InLineFSUtilsTest.java | 5 +-
.../common/fs/inline/TestInLineFileSystem.java | 6 +-
.../TestInLineFileSystemHFileInLiningBase.java | 0
.../TestInLineFileSystemWithHBaseHFileReader.java | 0
.../TestInLineFileSystemWithHFileReader.java | 0
.../common/fs/inline/TestInMemoryFileSystem.java | 0
.../hudi/common/fs/inline/TestParquetInLining.java | 0
.../common/functional/TestHoodieLogFormat.java | 4 +-
.../TestHoodieLogFormatAppendFailure.java | 0
.../hudi/common/model/TestHoodieFileGroup.java | 0
.../common/model/TestHoodiePartitionMetadata.java | 0
.../apache/hudi/common/model/TestHoodieRecord.java | 0
.../common/model/TestHoodieRecordDelegate.java | 0
.../hudi/common/table/TestHoodieTableConfig.java | 5 +-
.../common/table/TestHoodieTableMetaClient.java | 0
.../hudi/common/table/TestTableSchemaResolver.java | 7 +-
.../hudi/common/table/TestTimelineUtils.java | 0
.../hudi/common/table/log/TestLogReaderUtils.java | 2 +-
.../table/timeline/TestHoodieActiveTimeline.java | 2 +-
.../common/table/timeline/TestHoodieInstant.java | 0
.../view/TestHoodieTableFSViewWithClustering.java | 0
.../table/view/TestHoodieTableFileSystemView.java | 0
.../table/view/TestIncrementalFSViewSync.java | 0
.../TestRocksDBBasedIncrementalFSViewSync.java | 0
.../table/view/TestRocksDbBasedFileSystemView.java | 0
.../view/TestSpillableMapBasedFileSystemView.java | 0
...TestSpillableMapBasedIncrementalFSViewSync.java | 0
.../hudi/common/testutils/FileSystemTestUtils.java | 43 +-
.../common/testutils/HoodieCommonTestHarness.java | 0
.../hudi/common/testutils/HoodieTestTable.java | 43 +-
.../testutils/minicluster/HdfsTestService.java | 0
.../apache/hudi/common/util/TestAvroOrcUtils.java | 0
.../hudi/common/util/TestClusteringUtils.java | 0
.../apache/hudi/common/util/TestCommitUtils.java | 0
.../hudi/common/util/TestCompactionUtils.java | 3 +-
.../util/TestDFSPropertiesConfiguration.java | 9 +-
.../apache/hudi/common/util/TestFileIOUtils.java | 0
.../apache/hudi/common/util/TestMarkerUtils.java | 6 +-
.../apache/hudi/common/util/TestParquetUtils.java | 0
.../hudi/common/util/TestTablePathUtils.java | 6 +-
.../common/util/collection/TestBitCaskDiskMap.java | 0
.../util/collection/TestExternalSpillableMap.java | 0
.../util/collection/TestRocksDbBasedMap.java | 0
.../common/util/collection/TestRocksDbDiskMap.java | 0
.../TestFileBasedInternalSchemaStorageManager.java | 0
.../storage/TestHoodieAvroFileReaderFactory.java | 0
.../io/storage/TestHoodieBaseParquetWriter.java | 0
.../storage/TestHoodieHBaseHFileReaderWriter.java | 0
.../io/storage/TestHoodieHFileReaderWriter.java | 0
.../storage/TestHoodieHFileReaderWriterBase.java | 0
.../hudi/io/storage/TestHoodieOrcReaderWriter.java | 0
.../io/storage/TestHoodieReaderWriterBase.java | 0
.../TestFileSystemBackedTableMetadata.java | 0
.../hudi/metadata/TestHoodieMetadataPayload.java | 0
.../hudi/metadata/TestHoodieTableMetadataUtil.java | 0
.../resources/external-config/hudi-defaults.conf | 0
.../src/test/resources/props/testdfs.properties | 17 +
hudi-hadoop-mr/pom.xml | 22 +
.../TestHoodieMergeOnReadSnapshotReader.java | 5 +-
hudi-integ-test/pom.xml | 14 +
.../testsuite/writer/AvroFileDeltaInputWriter.java | 4 +-
.../apache/hudi/common}/fs/ConsistencyGuard.java | 2 +-
.../org/apache/hudi/storage/HoodieStorage.java | 8 +-
.../apache/hudi/storage/StorageConfiguration.java | 7 +
.../java/org/apache/hudi/storage/StoragePath.java | 7 +
.../apache/hudi/storage}/inline/InLineFSUtils.java | 60 +--
.../hudi/common/testutils/NetworkTestUtils.java | 0
.../hudi/io/storage/TestHoodieStorageBase.java | 4 +-
hudi-kafka-connect/pom.xml | 8 +
hudi-spark-datasource/hudi-spark-common/pom.xml | 26 +
hudi-spark-datasource/hudi-spark/pom.xml | 26 +
hudi-spark-datasource/hudi-spark2/pom.xml | 13 +
hudi-spark-datasource/hudi-spark3-common/pom.xml | 8 +
hudi-spark-datasource/hudi-spark3.0.x/pom.xml | 8 +
hudi-spark-datasource/hudi-spark3.1.x/pom.xml | 8 +
hudi-spark-datasource/hudi-spark3.2.x/pom.xml | 9 +
.../hudi-spark3.2plus-common/pom.xml | 8 +
hudi-spark-datasource/hudi-spark3.3.x/pom.xml | 16 +
hudi-spark-datasource/hudi-spark3.4.x/pom.xml | 9 +
hudi-spark-datasource/hudi-spark3.5.x/pom.xml | 9 +
hudi-sync/hudi-hive-sync/pom.xml | 27 +-
hudi-sync/hudi-sync-common/pom.xml | 14 +-
hudi-timeline-service/pom.xml | 13 +
hudi-utilities/pom.xml | 26 +
140 files changed, 856 insertions(+), 1021 deletions(-)
diff --git a/hudi-cli/pom.xml b/hudi-cli/pom.xml
index 8a6875a9df4..37408fd3ad2 100644
--- a/hudi-cli/pom.xml
+++ b/hudi-cli/pom.xml
@@ -169,6 +169,13 @@
<scope>test</scope>
<type>test-jar</type>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-client-common</artifactId>
diff --git a/hudi-client/hudi-client-common/pom.xml b/hudi-client/hudi-client-common/pom.xml
index 022f5d6faa0..48bc0ec9e0e 100644
--- a/hudi-client/hudi-client-common/pom.xml
+++ b/hudi-client/hudi-client-common/pom.xml
@@ -43,6 +43,16 @@
<artifactId>hudi-common</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-io</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-aws</artifactId>
@@ -111,6 +121,21 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-io</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-tests-common</artifactId>
diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java
index e9c9d39d216..58ea31bed21 100644
--- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java
+++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/HoodieTable.java
@@ -67,8 +67,8 @@ import org.apache.hudi.exception.HoodieInsertException;
import org.apache.hudi.exception.HoodieMetadataException;
import org.apache.hudi.exception.HoodieUpsertException;
import org.apache.hudi.exception.SchemaCompatibilityException;
-import org.apache.hudi.hadoop.fs.ConsistencyGuard;
-import org.apache.hudi.hadoop.fs.ConsistencyGuard.FileVisibility;
+import org.apache.hudi.common.fs.ConsistencyGuard;
+import org.apache.hudi.common.fs.ConsistencyGuard.FileVisibility;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.io.HoodieMergeHandle;
import org.apache.hudi.metadata.HoodieTableMetadata;
diff --git a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestInProcessLockProvider.java b/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestInProcessLockProvider.java
deleted file mode 100644
index c0e31b7e2bd..00000000000
--- a/hudi-client/hudi-client-common/src/test/java/org/apache/hudi/client/transaction/TestInProcessLockProvider.java
+++ /dev/null
@@ -1,527 +0,0 @@
-/*
- * 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.hudi.client.transaction;
-
-import org.apache.hudi.client.transaction.lock.InProcessLockProvider;
-import org.apache.hudi.common.config.LockConfiguration;
-import org.apache.hudi.common.config.TypedProperties;
-import org.apache.hudi.config.HoodieWriteConfig;
-import org.apache.hudi.exception.HoodieLockException;
-import org.apache.hudi.storage.StorageConfiguration;
-
-import junit.framework.AssertionFailedError;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-import static org.apache.hudi.common.testutils.HoodieTestUtils.getDefaultStorageConf;
-import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-
-public class TestInProcessLockProvider {
-
- private static final Logger LOG = LoggerFactory.getLogger(TestInProcessLockProvider.class);
- private final StorageConfiguration<?> storageConf = getDefaultStorageConf();
- private final LockConfiguration lockConfiguration1;
- private final LockConfiguration lockConfiguration2;
-
- public TestInProcessLockProvider() {
- TypedProperties properties = new TypedProperties();
- properties.put(HoodieWriteConfig.BASE_PATH.key(), "table1");
- lockConfiguration1 = new LockConfiguration(properties);
- properties.put(HoodieWriteConfig.BASE_PATH.key(), "table2");
- lockConfiguration2 = new LockConfiguration(properties);
- }
-
- @Test
- public void testLockIdentity() throws InterruptedException {
- // The lifecycle of an InProcessLockProvider should not affect the singleton lock
- // for a single table, i.e., all three writers should hold the same underlying lock instance
- // on the same table.
- // Writer 1: lock |----------------| unlock and close
- // Writer 2: try lock | ... lock |------| unlock and close
- // Writer 3: try lock | ... lock |------| unlock and close
- List<InProcessLockProvider> lockProviderList = new ArrayList<>();
- InProcessLockProvider lockProvider1 = new InProcessLockProvider(lockConfiguration1, storageConf);
- lockProviderList.add(lockProvider1);
- AtomicBoolean writer1Completed = new AtomicBoolean(false);
- AtomicBoolean writer2TryLock = new AtomicBoolean(false);
- AtomicBoolean writer2Locked = new AtomicBoolean(false);
- AtomicBoolean writer2Completed = new AtomicBoolean(false);
- AtomicBoolean writer3TryLock = new AtomicBoolean(false);
- AtomicBoolean writer3Completed = new AtomicBoolean(false);
-
- // Writer 1
- assertDoesNotThrow(() -> {
- LOG.info("Writer 1 tries to acquire the lock.");
- lockProvider1.lock();
- LOG.info("Writer 1 acquires the lock.");
- });
- // Writer 2 thread in parallel, should block
- // and later acquire the lock once it is released
- Thread writer2 = new Thread(() -> {
- InProcessLockProvider lockProvider2 = new InProcessLockProvider(lockConfiguration1, storageConf);
- lockProviderList.add(lockProvider2);
- assertDoesNotThrow(() -> {
- LOG.info("Writer 2 tries to acquire the lock.");
- writer2TryLock.set(true);
- lockProvider2.lock();
- LOG.info("Writer 2 acquires the lock.");
- });
- writer2Locked.set(true);
-
- while (!writer3TryLock.get()) {
- try {
- Thread.sleep(100);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- }
-
- assertDoesNotThrow(() -> {
- lockProvider2.unlock();
- LOG.info("Writer 2 releases the lock.");
- });
- lockProvider2.close();
- LOG.info("Writer 2 closes the lock provider.");
- writer2Completed.set(true);
- });
-
- Thread writer3 = new Thread(() -> {
- while (!writer2Locked.get() || !writer1Completed.get()) {
- try {
- Thread.sleep(10);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
- }
- // Lock instance of Writer 3 should be held by Writer 2
- InProcessLockProvider lockProvider3 = new InProcessLockProvider(lockConfiguration1, storageConf);
- lockProviderList.add(lockProvider3);
- boolean isLocked = lockProvider3.getLock().isWriteLocked();
- if (!isLocked) {
- writer3TryLock.set(true);
- throw new AssertionFailedError("The lock instance in Writer 3 should be held by Writer 2: "
- + lockProvider3.getLock());
- }
- assertDoesNotThrow(() -> {
- LOG.info("Writer 3 tries to acquire the lock.");
- writer3TryLock.set(true);
- lockProvider3.lock();
- LOG.info("Writer 3 acquires the lock.");
- });
-
- assertDoesNotThrow(() -> {
- lockProvider3.unlock();
- LOG.info("Writer 3 releases the lock.");
- });
- lockProvider3.close();
- LOG.info("Writer 3 closes the lock provider.");
- writer3Completed.set(true);
- });
-
- writer2.start();
- writer3.start();
-
- while (!writer2TryLock.get()) {
- Thread.sleep(100);
- }
-
- assertDoesNotThrow(() -> {
- lockProvider1.unlock();
- LOG.info("Writer 1 releases the lock.");
- lockProvider1.close();
- LOG.info("Writer 1 closes the lock provider.");
- writer1Completed.set(true);
- });
-
- try {
- writer2.join();
- writer3.join();
- } catch (InterruptedException e) {
- // Ignore any exception
- }
- Assertions.assertTrue(writer2Completed.get());
- Assertions.assertTrue(writer3Completed.get());
- Assertions.assertEquals(lockProviderList.get(0).getLock(), lockProviderList.get(1).getLock());
- Assertions.assertEquals(lockProviderList.get(1).getLock(), lockProviderList.get(2).getLock());
-
- writer2.interrupt();
- writer3.interrupt();
- }
-
- @Test
- public void testLockAcquisition() {
- InProcessLockProvider inProcessLockProvider = new InProcessLockProvider(lockConfiguration1, storageConf);
- assertDoesNotThrow(() -> {
- inProcessLockProvider.lock();
- });
- assertDoesNotThrow(() -> {
- inProcessLockProvider.unlock();
- });
- }
-
- @Test
- public void testLockReAcquisitionBySameThread() {
- InProcessLockProvider inProcessLockProvider = new InProcessLockProvider(lockConfiguration1, storageConf);
- assertDoesNotThrow(() -> {
- inProcessLockProvider.lock();
- });
- assertThrows(HoodieLockException.class, () -> {
- inProcessLockProvider.lock();
- });
- assertDoesNotThrow(() -> {
- inProcessLockProvider.unlock();
- });
- }
-
- @Test
- public void testLockReAcquisitionBySameThreadWithTwoTables() {
- InProcessLockProvider inProcessLockProvider1 = new InProcessLockProvider(lockConfiguration1, storageConf);
- InProcessLockProvider inProcessLockProvider2 = new InProcessLockProvider(lockConfiguration2, storageConf);
-
- assertDoesNotThrow(() -> {
- inProcessLockProvider1.lock();
- });
- assertDoesNotThrow(() -> {
- inProcessLockProvider2.lock();
- });
- assertThrows(HoodieLockException.class, () -> {
- inProcessLockProvider2.lock();
- });
- assertThrows(HoodieLockException.class, () -> {
- inProcessLockProvider1.lock();
- });
- assertDoesNotThrow(() -> {
- inProcessLockProvider1.unlock();
- });
- assertDoesNotThrow(() -> {
- inProcessLockProvider2.unlock();
- });
- }
-
- @Test
- public void testLockReAcquisitionByDifferentThread() {
- InProcessLockProvider inProcessLockProvider = new InProcessLockProvider(lockConfiguration1, storageConf);
- final AtomicBoolean writer2Completed = new AtomicBoolean(false);
-
- // Main test thread
- assertDoesNotThrow(() -> {
- inProcessLockProvider.lock();
- });
-
- // Another writer thread in parallel, should block
- // and later acquire the lock once it is released
- Thread writer2 = new Thread(new Runnable() {
- @Override
- public void run() {
- assertDoesNotThrow(() -> {
- inProcessLockProvider.lock();
- });
- assertDoesNotThrow(() -> {
- inProcessLockProvider.unlock();
- });
- writer2Completed.set(true);
- }
- });
- writer2.start();
-
- assertDoesNotThrow(() -> {
- inProcessLockProvider.unlock();
- });
-
- try {
- writer2.join();
- } catch (InterruptedException e) {
- //
- }
- Assertions.assertTrue(writer2Completed.get());
-
- writer2.interrupt();
- }
-
- @Test
- public void testLockReAcquisitionByDifferentThreadWithTwoTables() {
- InProcessLockProvider inProcessLockProvider1 = new InProcessLockProvider(lockConfiguration1, storageConf);
- InProcessLockProvider inProcessLockProvider2 = new InProcessLockProvider(lockConfiguration2, storageConf);
-
- final AtomicBoolean writer2Stream1Completed = new AtomicBoolean(false);
- final AtomicBoolean writer2Stream2Completed = new AtomicBoolean(false);
-
- // Main test thread
- assertDoesNotThrow(() -> {
- inProcessLockProvider1.lock();
- });
- assertDoesNotThrow(() -> {
- inProcessLockProvider2.lock();
- });
-
- // Another writer thread in parallel, should block
- // and later acquire the lock once it is released
- Thread writer2Stream1 = new Thread(new Runnable() {
- @Override
- public void run() {
- assertDoesNotThrow(() -> {
- inProcessLockProvider1.lock();
- });
- assertDoesNotThrow(() -> {
- inProcessLockProvider1.unlock();
- });
- writer2Stream1Completed.set(true);
- }
- });
- Thread writer2Stream2 = new Thread(new Runnable() {
- @Override
- public void run() {
- assertDoesNotThrow(() -> {
- inProcessLockProvider2.lock();
- });
- assertDoesNotThrow(() -> {
- inProcessLockProvider2.unlock();
- });
- writer2Stream2Completed.set(true);
- }
- });
-
- writer2Stream1.start();
- writer2Stream2.start();
-
- assertDoesNotThrow(() -> {
- inProcessLockProvider1.unlock();
- });
- assertDoesNotThrow(() -> {
- inProcessLockProvider2.unlock();
- });
-
- try {
- writer2Stream1.join();
- writer2Stream2.join();
- } catch (InterruptedException e) {
- //
- }
- Assertions.assertTrue(writer2Stream1Completed.get());
- Assertions.assertTrue(writer2Stream2Completed.get());
-
- writer2Stream1.interrupt();
- writer2Stream2.interrupt();
- }
-
- @Test
- public void testTryLockAcquisition() {
- InProcessLockProvider inProcessLockProvider = new InProcessLockProvider(lockConfiguration1, storageConf);
- Assertions.assertTrue(inProcessLockProvider.tryLock());
- assertDoesNotThrow(() -> {
- inProcessLockProvider.unlock();
- });
- }
-
- @Test
- public void testTryLockAcquisitionWithTimeout() {
- InProcessLockProvider inProcessLockProvider = new InProcessLockProvider(lockConfiguration1, storageConf);
- Assertions.assertTrue(inProcessLockProvider.tryLock(1, TimeUnit.MILLISECONDS));
- assertDoesNotThrow(() -> {
- inProcessLockProvider.unlock();
- });
- }
-
- @Test
- public void testTryLockReAcquisitionBySameThread() {
- InProcessLockProvider inProcessLockProvider = new InProcessLockProvider(lockConfiguration1, storageConf);
- Assertions.assertTrue(inProcessLockProvider.tryLock());
- assertThrows(HoodieLockException.class, () -> {
- inProcessLockProvider.tryLock(1, TimeUnit.MILLISECONDS);
- });
- assertDoesNotThrow(() -> {
- inProcessLockProvider.unlock();
- });
- }
-
- @Test
- public void testTryLockReAcquisitionByDifferentThread() {
- InProcessLockProvider inProcessLockProvider = new InProcessLockProvider(lockConfiguration1, storageConf);
- final AtomicBoolean writer2Completed = new AtomicBoolean(false);
-
- // Main test thread
- Assertions.assertTrue(inProcessLockProvider.tryLock());
-
- // Another writer thread
- Thread writer2 = new Thread(() -> {
- Assertions.assertFalse(inProcessLockProvider.tryLock(100L, TimeUnit.MILLISECONDS));
- writer2Completed.set(true);
- });
- writer2.start();
- try {
- writer2.join();
- } catch (InterruptedException e) {
- //
- }
-
- Assertions.assertTrue(writer2Completed.get());
- assertDoesNotThrow(() -> {
- inProcessLockProvider.unlock();
- });
-
- writer2.interrupt();
- }
-
- @Test
- public void testTryUnLockByDifferentThread() {
- InProcessLockProvider inProcessLockProvider = new InProcessLockProvider(lockConfiguration1, storageConf);
- final AtomicBoolean writer3Completed = new AtomicBoolean(false);
-
- // Main test thread
- Assertions.assertTrue(inProcessLockProvider.tryLock());
-
- // Another writer thread
- Thread writer2 = new Thread(() -> {
- assertDoesNotThrow(() -> {
- inProcessLockProvider.unlock();
- });
- });
- writer2.start();
- try {
- writer2.join();
- } catch (InterruptedException e) {
- //
- }
-
- // try acquiring by diff thread. should fail. since main thread still have acquired the lock. if previous unblock by a different thread would have succeeded, this lock
- // acquisition would succeed.
- Thread writer3 = new Thread(() -> {
- Assertions.assertFalse(inProcessLockProvider.tryLock(50, TimeUnit.MILLISECONDS));
- writer3Completed.set(true);
- });
- writer3.start();
- try {
- writer3.join();
- } catch (InterruptedException e) {
- //
- }
-
- Assertions.assertTrue(writer3Completed.get());
- assertDoesNotThrow(() -> {
- // unlock by main thread should succeed.
- inProcessLockProvider.unlock();
- });
-
- writer2.interrupt();
- writer3.interrupt();
- }
-
- @Test
- public void testTryLockAcquisitionBeforeTimeOutFromTwoThreads() {
- final InProcessLockProvider inProcessLockProvider = new InProcessLockProvider(lockConfiguration1, storageConf);
- final int threadCount = 3;
- final long awaitMaxTimeoutMs = 2000L;
- final CountDownLatch latch = new CountDownLatch(threadCount);
- final AtomicBoolean writer1Completed = new AtomicBoolean(false);
- final AtomicBoolean writer2Completed = new AtomicBoolean(false);
-
- // Let writer1 get the lock first, then wait for others
- // to join the sync up point.
- Thread writer1 = new Thread(() -> {
- Assertions.assertTrue(inProcessLockProvider.tryLock());
- latch.countDown();
- try {
- latch.await(awaitMaxTimeoutMs, TimeUnit.MILLISECONDS);
- // Following sleep is to make sure writer2 attempts
- // to try lock and to get bocked on the lock which
- // this thread is currently holding.
- Thread.sleep(50);
- } catch (InterruptedException e) {
- //
- }
- assertDoesNotThrow(() -> {
- inProcessLockProvider.unlock();
- });
- writer1Completed.set(true);
- });
- writer1.start();
-
- // Writer2 will block on trying to acquire the lock
- // and will eventually get the lock before the timeout.
- Thread writer2 = new Thread(() -> {
- latch.countDown();
- Assertions.assertTrue(inProcessLockProvider.tryLock(awaitMaxTimeoutMs, TimeUnit.MILLISECONDS));
- assertDoesNotThrow(() -> {
- inProcessLockProvider.unlock();
- });
- writer2Completed.set(true);
- });
- writer2.start();
-
- // Let writer1 and writer2 wait at the sync up
- // point to make sure they run in parallel and
- // one get blocked by the other.
- latch.countDown();
- try {
- writer1.join();
- writer2.join();
- } catch (InterruptedException e) {
- //
- }
-
- // Make sure both writers actually completed good
- Assertions.assertTrue(writer1Completed.get());
- Assertions.assertTrue(writer2Completed.get());
-
- writer1.interrupt();
- writer2.interrupt();
- }
-
- @Test
- public void testLockReleaseByClose() {
- InProcessLockProvider inProcessLockProvider = new InProcessLockProvider(lockConfiguration1, storageConf);
- assertDoesNotThrow(() -> {
- inProcessLockProvider.lock();
- });
- assertDoesNotThrow(() -> {
- inProcessLockProvider.close();
- });
- }
-
- @Test
- public void testRedundantUnlock() {
- InProcessLockProvider inProcessLockProvider = new InProcessLockProvider(lockConfiguration1, storageConf);
- assertDoesNotThrow(() -> {
- inProcessLockProvider.lock();
- });
- assertDoesNotThrow(() -> {
- inProcessLockProvider.unlock();
- });
- assertDoesNotThrow(() -> {
- inProcessLockProvider.unlock();
- });
- }
-
- @Test
- public void testUnlockWithoutLock() {
- InProcessLockProvider inProcessLockProvider = new InProcessLockProvider(lockConfiguration1, storageConf);
- assertDoesNotThrow(() -> {
- inProcessLockProvider.unlock();
- });
- }
-}
diff --git a/hudi-client/hudi-flink-client/pom.xml b/hudi-client/hudi-flink-client/pom.xml
index db06ab867fc..c2e1d883aba 100644
--- a/hudi-client/hudi-flink-client/pom.xml
+++ b/hudi-client/hudi-flink-client/pom.xml
@@ -55,6 +55,11 @@
<artifactId>hudi-client-common</artifactId>
<version>${project.parent.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<!-- Flink -->
<dependency>
@@ -132,6 +137,14 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-hadoop-mr</artifactId>
diff --git a/hudi-client/hudi-java-client/pom.xml b/hudi-client/hudi-java-client/pom.xml
index 46829b19b5e..b4de6e103dd 100644
--- a/hudi-client/hudi-java-client/pom.xml
+++ b/hudi-client/hudi-java-client/pom.xml
@@ -37,6 +37,11 @@
</dependency>
<!-- Hudi -->
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-client-common</artifactId>
@@ -64,6 +69,14 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-client-common</artifactId>
diff --git a/hudi-client/hudi-spark-client/pom.xml b/hudi-client/hudi-spark-client/pom.xml
index d70ecedefee..85a102954e4 100644
--- a/hudi-client/hudi-spark-client/pom.xml
+++ b/hudi-client/hudi-spark-client/pom.xml
@@ -55,6 +55,18 @@
<version>${project.parent.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-io</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
<!-- Spark -->
<dependency>
<groupId>org.apache.spark</groupId>
@@ -90,6 +102,22 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-io</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-client-common</artifactId>
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestMultiFS.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestMultiFS.java
index 1bb4b9ff70e..2d29e936d15 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestMultiFS.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/client/TestMultiFS.java
@@ -32,6 +32,7 @@ import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.index.HoodieIndex;
import org.apache.hudi.keygen.constant.KeyGeneratorOptions;
+import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.testutils.HoodieClientTestUtils;
import org.apache.hudi.testutils.HoodieSparkClientTestHarness;
@@ -133,7 +134,7 @@ public class TestMultiFS extends HoodieSparkClientTestHarness {
// Read from hdfs
FileSystem fs = HadoopFSUtils.getFs(dfsBasePath, HoodieTestUtils.getDefaultStorageConf());
- HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(fs.getConf(), dfsBasePath);
+ HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(HoodieStorageUtils.getStorageConf(fs.getConf()), dfsBasePath);
HoodieTimeline timeline = new HoodieActiveTimeline(metaClient).getCommitTimeline();
Dataset<Row> readRecords = HoodieClientTestUtils.readCommit(dfsBasePath, sqlContext, timeline, readCommitTime);
assertEquals(readRecords.count(), records.size());
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/TestConsistencyGuard.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/TestConsistencyGuard.java
index 072b88b1f6c..9461e781900 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/TestConsistencyGuard.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/TestConsistencyGuard.java
@@ -23,7 +23,7 @@ import org.apache.hudi.common.fs.FailSafeConsistencyGuard;
import org.apache.hudi.common.fs.OptimisticConsistencyGuard;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.testutils.FileCreateUtils;
-import org.apache.hudi.hadoop.fs.ConsistencyGuard;
+import org.apache.hudi.common.fs.ConsistencyGuard;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.testutils.HoodieSparkClientTestHarness;
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/marker/TestDirectWriteMarkers.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/marker/TestDirectWriteMarkers.java
index ac80e61db28..ec503a8774c 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/marker/TestDirectWriteMarkers.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/marker/TestDirectWriteMarkers.java
@@ -19,7 +19,7 @@
package org.apache.hudi.table.marker;
import org.apache.hudi.client.common.HoodieSparkEngineContext;
-import org.apache.hudi.common.testutils.FileSystemTestUtils;
+import org.apache.hudi.common.testutils.HoodieTestTable;
import org.apache.hudi.common.util.CollectionUtils;
import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StoragePath;
@@ -61,7 +61,7 @@ public class TestDirectWriteMarkers extends TestWriteMarkersBase {
@Override
void verifyMarkersInFileSystem(boolean isTablePartitioned) throws IOException {
- List<StoragePathInfo> markerFiles = FileSystemTestUtils.listRecursive(storage, markerFolderPath)
+ List<StoragePathInfo> markerFiles = HoodieTestTable.listRecursive(storage, markerFolderPath)
.stream().filter(status -> status.getPath().getName().contains(".marker"))
.sorted().collect(Collectors.toList());
assertEquals(3, markerFiles.size());
diff --git a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/marker/TestWriteMarkersBase.java b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/marker/TestWriteMarkersBase.java
index 7eba0f31ca8..040ac5f59b1 100644
--- a/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/marker/TestWriteMarkersBase.java
+++ b/hudi-client/hudi-spark-client/src/test/java/org/apache/hudi/table/marker/TestWriteMarkersBase.java
@@ -21,7 +21,6 @@ package org.apache.hudi.table.marker;
import org.apache.hudi.client.common.HoodieSparkEngineContext;
import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.IOType;
-import org.apache.hudi.common.testutils.FileSystemTestUtils;
import org.apache.hudi.common.testutils.HoodieCommonTestHarness;
import org.apache.hudi.common.util.CollectionUtils;
import org.apache.hudi.common.util.MarkerUtils;
@@ -38,6 +37,7 @@ import java.io.IOException;
import java.util.List;
import java.util.stream.Collectors;
+import static org.apache.hudi.common.testutils.HoodieTestTable.listRecursive;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertIterableEquals;
@@ -107,7 +107,7 @@ public abstract class TestWriteMarkersBase extends HoodieCommonTestHarness {
createSomeMarkers(isTablePartitioned);
// add invalid file
createInvalidFile(isTablePartitioned ? "2020/06/01" : "", "invalid_file3");
- long fileSize = FileSystemTestUtils.listRecursive(storage, markerFolderPath).stream()
+ long fileSize = listRecursive(storage, markerFolderPath).stream()
.filter(fileStatus -> !fileStatus.getPath().getName().contains(MarkerUtils.MARKER_TYPE_FILENAME))
.count();
assertEquals(fileSize, 4);
@@ -128,7 +128,7 @@ public abstract class TestWriteMarkersBase extends HoodieCommonTestHarness {
createSomeMarkers(isTablePartitioned);
// add invalid file
createInvalidFile(isTablePartitioned ? "2020/06/01" : "", "invalid_file3");
- long fileSize = FileSystemTestUtils.listRecursive(storage, markerFolderPath).stream()
+ long fileSize = listRecursive(storage, markerFolderPath).stream()
.filter(fileStatus -> !fileStatus.getPath().getName().contains(MarkerUtils.MARKER_TYPE_FILENAME))
.count();
assertEquals(fileSize, 4);
diff --git a/hudi-common/pom.xml b/hudi-common/pom.xml
index 6e2aee560f4..b02acb8d69b 100644
--- a/hudi-common/pom.xml
+++ b/hudi-common/pom.xml
@@ -109,12 +109,6 @@
<version>${project.version}</version>
</dependency>
- <dependency>
- <groupId>org.apache.hudi</groupId>
- <artifactId>hudi-hadoop-common</artifactId>
- <version>${project.version}</version>
- </dependency>
-
<dependency>
<groupId>org.openjdk.jol</groupId>
<artifactId>jol-core</artifactId>
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/bootstrap/index/HFileBootstrapIndex.java b/hudi-common/src/main/java/org/apache/hudi/common/bootstrap/index/HFileBootstrapIndex.java
index 693eb7b6719..a1c6e7901b2 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/bootstrap/index/HFileBootstrapIndex.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/bootstrap/index/HFileBootstrapIndex.java
@@ -242,7 +242,7 @@ public class HFileBootstrapIndex extends BootstrapIndex {
LOG.info("Opening HFile for reading :" + hFilePath);
StoragePath path = new StoragePath(hFilePath);
long fileSize = storage.getPathInfo(path).getLength();
- SeekableDataInputStream stream = storage.openSeekable(path);
+ SeekableDataInputStream stream = storage.openSeekable(path, true);
return new HFileReaderImpl(stream, fileSize);
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieConfig.java b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieConfig.java
index f21721391d2..d81f4fa19d4 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieConfig.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/config/HoodieConfig.java
@@ -33,6 +33,7 @@ import java.util.List;
import java.util.Properties;
import static org.apache.hudi.common.util.ConfigUtils.getRawValueWithAltKeys;
+import static org.apache.hudi.common.util.ConfigUtils.loadGlobalProperties;
/**
* This class deals with {@link ConfigProperty} and provides get/set functionalities.
@@ -234,7 +235,7 @@ public class HoodieConfig implements Serializable {
public TypedProperties getProps(boolean includeGlobalProps) {
if (includeGlobalProps) {
- TypedProperties mergedProps = DFSPropertiesConfiguration.getGlobalProps();
+ TypedProperties mergedProps = loadGlobalProperties();
mergedProps.putAll(props);
return mergedProps;
} else {
diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/NoOpConsistencyGuard.java b/hudi-common/src/main/java/org/apache/hudi/common/config/PropertiesConfig.java
similarity index 56%
copy from hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/NoOpConsistencyGuard.java
copy to hudi-common/src/main/java/org/apache/hudi/common/config/PropertiesConfig.java
index 1f8401a0b88..6edbe5bb560 100644
--- a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/NoOpConsistencyGuard.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/config/PropertiesConfig.java
@@ -17,33 +17,16 @@
* under the License.
*/
-package org.apache.hudi.hadoop.fs;
-
-import org.apache.hudi.storage.StoragePath;
-
-import java.util.List;
+package org.apache.hudi.common.config;
/**
- * Default Consistency guard that does nothing. Used for lake storage which provided read-after-write
- * guarantees.
+ * Used for loading filesystem specific configs
*/
-public class NoOpConsistencyGuard implements ConsistencyGuard {
-
- @Override
- public void waitTillFileAppears(StoragePath filePath) {
- }
-
- @Override
- public void waitTillFileDisappears(StoragePath filePath) {
- }
-
- @Override
- public void waitTillAllFilesAppear(String dirPath, List<String> files) {
-
- }
-
- @Override
- public void waitTillAllFilesDisappear(String dirPath, List<String> files) {
-
+public abstract class PropertiesConfig {
+ /**
+ * return any global properties for the filesystem
+ */
+ public TypedProperties getGlobalProperties() {
+ return new TypedProperties();
}
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java
index 1c24840499e..b2f87b9f01a 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/fs/FSUtils.java
@@ -33,11 +33,6 @@ import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.exception.HoodieValidationException;
import org.apache.hudi.exception.InvalidHoodiePathException;
-import org.apache.hudi.hadoop.fs.CachingPath;
-import org.apache.hudi.hadoop.fs.HadoopFSUtils;
-import org.apache.hudi.hadoop.fs.HoodieWrapperFileSystem;
-import org.apache.hudi.hadoop.fs.inline.InLineFSUtils;
-import org.apache.hudi.hadoop.fs.inline.InLineFileSystem;
import org.apache.hudi.metadata.HoodieTableMetadata;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StorageConfiguration;
@@ -45,6 +40,7 @@ import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.storage.StoragePathFilter;
import org.apache.hudi.storage.StoragePathInfo;
import org.apache.hudi.storage.StorageSchemes;
+import org.apache.hudi.storage.inline.InLineFSUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
@@ -75,6 +71,8 @@ import java.util.regex.Pattern;
import java.util.stream.Collectors;
import java.util.stream.Stream;
+import static org.apache.hudi.storage.HoodieStorageUtils.getStorageConfWithCopy;
+
/**
* Utility functions related to accessing the file storage.
*/
@@ -93,20 +91,6 @@ public class FSUtils {
private static final StoragePathFilter ALLOW_ALL_FILTER = file -> true;
- public static Configuration buildInlineConf(Configuration conf) {
- Configuration inlineConf = new Configuration(conf);
- inlineConf.set("fs." + InLineFileSystem.SCHEME + ".impl", InLineFileSystem.class.getName());
- inlineConf.setClassLoader(InLineFileSystem.class.getClassLoader());
- return inlineConf;
- }
-
- public static StorageConfiguration<?> buildInlineConf(StorageConfiguration<?> storageConf) {
- StorageConfiguration<?> inlineConf = storageConf.newInstance();
- inlineConf.set("fs." + InLineFileSystem.SCHEME + ".impl", InLineFileSystem.class.getName());
- (inlineConf.unwrapAs(Configuration.class)).setClassLoader(InLineFileSystem.class.getClassLoader());
- return inlineConf;
- }
-
/**
* Check if table already exists in the given path.
*
@@ -208,21 +192,7 @@ public class FSUtils {
* Given a base partition and a partition path, return relative path of partition path to the base path.
*/
public static String getRelativePartitionPath(Path basePath, Path fullPartitionPath) {
- basePath = CachingPath.getPathWithoutSchemeAndAuthority(basePath);
- fullPartitionPath = CachingPath.getPathWithoutSchemeAndAuthority(fullPartitionPath);
-
- String fullPartitionPathStr = fullPartitionPath.toString();
-
- if (!fullPartitionPathStr.startsWith(basePath.toString())) {
- throw new IllegalArgumentException("Partition path \"" + fullPartitionPathStr
- + "\" does not belong to base-path \"" + basePath + "\"");
- }
-
- int partitionStartIndex = fullPartitionPathStr.indexOf(basePath.getName(),
- basePath.getParent() == null ? 0 : basePath.getParent().toString().length());
- // Partition-Path could be empty for non-partitioned tables
- return partitionStartIndex + basePath.getName().length() == fullPartitionPathStr.length() ? ""
- : fullPartitionPathStr.substring(partitionStartIndex + basePath.getName().length() + 1);
+ return getRelativePartitionPath(new StoragePath(basePath.toUri()), new StoragePath(fullPartitionPath.toUri()));
}
public static String getRelativePartitionPath(StoragePath basePath, StoragePath fullPartitionPath) {
@@ -548,12 +518,12 @@ public class FSUtils {
public static boolean isLogFile(StoragePath logPath) {
String scheme = logPath.toUri().getScheme();
- return isLogFile(InLineFileSystem.SCHEME.equals(scheme)
+ return isLogFile(InLineFSUtils.SCHEME.equals(scheme)
? InLineFSUtils.getOuterFilePathFromInlinePath(logPath).getName() : logPath.getName());
}
public static boolean isLogFile(Path logPath) {
- return isLogFile(logPath.getName());
+ return isLogFile(new StoragePath(logPath.getName()));
}
public static boolean isLogFile(String fileName) {
@@ -705,16 +675,7 @@ public class FSUtils {
}
public static Path constructAbsolutePathInHadoopPath(String basePath, String relativePartitionPath) {
- if (StringUtils.isNullOrEmpty(relativePartitionPath)) {
- return new Path(basePath);
- }
-
- // NOTE: We have to chop leading "/" to make sure Hadoop does not treat it like
- // absolute path
- String properPartitionPath = relativePartitionPath.startsWith(PATH_SEPARATOR)
- ? relativePartitionPath.substring(1)
- : relativePartitionPath;
- return constructAbsolutePath(new CachingPath(basePath), properPartitionPath);
+ return new Path(constructAbsolutePath(basePath, relativePartitionPath).toUri());
}
public static StoragePath constructAbsolutePath(String basePath, String relativePartitionPath) {
@@ -730,11 +691,6 @@ public class FSUtils {
return constructAbsolutePath(new StoragePath(basePath), properPartitionPath);
}
- public static Path constructAbsolutePath(Path basePath, String relativePartitionPath) {
- // For non-partitioned table, return only base-path
- return StringUtils.isNullOrEmpty(relativePartitionPath) ? basePath : new CachingPath(basePath, relativePartitionPath);
- }
-
public static StoragePath constructAbsolutePath(StoragePath basePath, String relativePartitionPath) {
// For non-partitioned table, return only base-path
return StringUtils.isNullOrEmpty(relativePartitionPath) ? basePath : new StoragePath(basePath, relativePartitionPath);
@@ -765,14 +721,6 @@ public class FSUtils {
return fs.getUri() + fullPartitionPath.toUri().getRawPath();
}
- public static Configuration registerFileSystem(StoragePath file, Configuration conf) {
- Configuration returnConf = new Configuration(conf);
- String scheme = HadoopFSUtils.getFs(file.toString(), conf).getScheme();
- returnConf.set("fs." + HoodieWrapperFileSystem.getHoodieScheme(scheme) + ".impl",
- HoodieWrapperFileSystem.class.getName());
- return returnConf;
- }
-
/**
* Helper to filter out paths under metadata folder when running fs.globStatus.
*
@@ -826,7 +774,7 @@ public class FSUtils {
List<String> subPaths) {
Map<String, T> result = new HashMap<>();
if (subPaths.size() > 0) {
- StorageConfiguration<Configuration> conf = HadoopFSUtils.getStorageConfWithCopy(fs.getConf());
+ StorageConfiguration<Configuration> conf = getStorageConfWithCopy(fs.getConf());
int actualParallelism = Math.min(subPaths.size(), parallelism);
hoodieEngineContext.setJobStatus(FSUtils.class.getSimpleName(),
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/fs/FailSafeConsistencyGuard.java b/hudi-common/src/main/java/org/apache/hudi/common/fs/FailSafeConsistencyGuard.java
index decd1099dac..f873d8d22a5 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/fs/FailSafeConsistencyGuard.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/fs/FailSafeConsistencyGuard.java
@@ -19,7 +19,6 @@
package org.apache.hudi.common.fs;
import org.apache.hudi.common.util.ValidationUtils;
-import org.apache.hudi.hadoop.fs.ConsistencyGuard;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.storage.StoragePathInfo;
diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/NoOpConsistencyGuard.java b/hudi-common/src/main/java/org/apache/hudi/common/fs/NoOpConsistencyGuard.java
similarity index 97%
rename from hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/NoOpConsistencyGuard.java
rename to hudi-common/src/main/java/org/apache/hudi/common/fs/NoOpConsistencyGuard.java
index 1f8401a0b88..f47a180b856 100644
--- a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/NoOpConsistencyGuard.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/fs/NoOpConsistencyGuard.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.hudi.hadoop.fs;
+package org.apache.hudi.common.fs;
import org.apache.hudi.storage.StoragePath;
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/fs/OptimisticConsistencyGuard.java b/hudi-common/src/main/java/org/apache/hudi/common/fs/OptimisticConsistencyGuard.java
index 8e0f9a0dc41..dfe7047999f 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/fs/OptimisticConsistencyGuard.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/fs/OptimisticConsistencyGuard.java
@@ -18,7 +18,6 @@
package org.apache.hudi.common.fs;
-import org.apache.hudi.hadoop.fs.ConsistencyGuard;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.storage.HoodieStorage;
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
index f694d7cefc8..bedf0204bf8 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/HoodieTableMetaClient.java
@@ -21,9 +21,11 @@ package org.apache.hudi.common.table;
import org.apache.hudi.common.config.ConfigProperty;
import org.apache.hudi.common.config.HoodieConfig;
import org.apache.hudi.common.config.HoodieMetaserverConfig;
+import org.apache.hudi.common.fs.ConsistencyGuard;
import org.apache.hudi.common.fs.ConsistencyGuardConfig;
import org.apache.hudi.common.fs.FailSafeConsistencyGuard;
import org.apache.hudi.common.fs.FileSystemRetryConfig;
+import org.apache.hudi.common.fs.NoOpConsistencyGuard;
import org.apache.hudi.common.model.HoodieRecordPayload;
import org.apache.hudi.common.model.HoodieTableType;
import org.apache.hudi.common.model.HoodieTimelineTimeZone;
@@ -40,8 +42,6 @@ import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.ValidationUtils;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.TableNotFoundException;
-import org.apache.hudi.hadoop.fs.ConsistencyGuard;
-import org.apache.hudi.hadoop.fs.NoOpConsistencyGuard;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StorageConfiguration;
@@ -65,7 +65,6 @@ import java.util.stream.Stream;
import static org.apache.hudi.common.util.ConfigUtils.containsConfigProperty;
import static org.apache.hudi.common.util.ConfigUtils.getStringWithAltKeys;
-import static org.apache.hudi.hadoop.fs.HadoopFSUtils.getStorageWithWrapperFS;
/**
* <code>HoodieTableMetaClient</code> allows to access meta-data about a hoodie table It returns meta-data about
@@ -303,7 +302,7 @@ public class HoodieTableMetaClient implements Serializable {
consistencyGuardConfig)
: new NoOpConsistencyGuard();
- storage = getStorageWithWrapperFS(
+ storage = HoodieStorageUtils.getStorage(
metaPath,
getStorageConf(),
fileSystemRetryConfig.isFileSystemActionRetryEnable(),
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFileReader.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFileReader.java
index c8bddc1d66c..98227ef1074 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFileReader.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogFileReader.java
@@ -474,7 +474,7 @@ public class HoodieLogFileReader implements HoodieLogFormat.Reader {
HoodieLogFile logFile,
int bufferSize) {
try {
- return storage.openSeekable(logFile.getPath(), bufferSize);
+ return storage.openSeekable(logFile.getPath(), bufferSize, true);
} catch (IOException e) {
throw new HoodieIOException("Unable to get seekable input stream for " + logFile, e);
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieHFileDataBlock.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieHFileDataBlock.java
index eace77bad8b..50c5e4af6e3 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieHFileDataBlock.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieHFileDataBlock.java
@@ -21,18 +21,14 @@ package org.apache.hudi.common.table.log.block;
import org.apache.hudi.avro.HoodieAvroUtils;
import org.apache.hudi.common.config.HoodieConfig;
import org.apache.hudi.common.config.HoodieReaderConfig;
-import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieFileFormat;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType;
-import org.apache.hudi.common.table.log.block.HoodieLogBlock.HeaderMetadataType;
-import org.apache.hudi.common.table.log.block.HoodieLogBlock.HoodieLogBlockContentLocation;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.ClosableIterator;
import org.apache.hudi.common.util.collection.CloseableMappingIterator;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieIOException;
-import org.apache.hudi.hadoop.fs.inline.InLineFSUtils;
import org.apache.hudi.io.SeekableDataInputStream;
import org.apache.hudi.io.storage.HoodieAvroHFileReaderImplBase;
import org.apache.hudi.io.storage.HoodieFileReader;
@@ -43,6 +39,7 @@ import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StorageConfiguration;
import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.inline.InLineFSUtils;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
@@ -191,8 +188,7 @@ public class HoodieHFileDataBlock extends HoodieDataBlock {
protected <T> ClosableIterator<HoodieRecord<T>> deserializeRecords(byte[] content, HoodieRecordType type) throws IOException {
checkState(readerSchema != null, "Reader's schema has to be non-null");
- StorageConfiguration<?> storageConf =
- FSUtils.buildInlineConf(getBlockContentLocation().get().getStorageConf());
+ StorageConfiguration<?> storageConf = getBlockContentLocation().get().getStorageConf().getInline();
HoodieStorage storage = HoodieStorageUtils.getStorage(pathForReader, storageConf);
// Read the content
try (HoodieFileReader reader =
@@ -211,7 +207,7 @@ public class HoodieHFileDataBlock extends HoodieDataBlock {
// NOTE: It's important to extend Hadoop configuration here to make sure configuration
// is appropriately carried over
- StorageConfiguration<?> inlineConf = FSUtils.buildInlineConf(blockContentLoc.getStorageConf());
+ StorageConfiguration<?> inlineConf = blockContentLoc.getStorageConf().getInline();
StoragePath inlinePath = InLineFSUtils.getInlineFilePath(
blockContentLoc.getLogFile().getPath(),
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieParquetDataBlock.java b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieParquetDataBlock.java
index dc1dd4063aa..d426480fc68 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieParquetDataBlock.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieParquetDataBlock.java
@@ -19,20 +19,19 @@
package org.apache.hudi.common.table.log.block;
import org.apache.hudi.common.config.HoodieConfig;
-import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieFileFormat;
import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType;
-import org.apache.hudi.common.util.collection.ClosableIterator;
import org.apache.hudi.common.util.Option;
-import org.apache.hudi.hadoop.fs.HadoopFSUtils;
-import org.apache.hudi.hadoop.fs.inline.InLineFSUtils;
+import org.apache.hudi.common.util.collection.ClosableIterator;
import org.apache.hudi.io.SeekableDataInputStream;
import org.apache.hudi.io.storage.HoodieFileReaderFactory;
import org.apache.hudi.io.storage.HoodieFileWriter;
import org.apache.hudi.io.storage.HoodieFileWriterFactory;
+import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StorageConfiguration;
import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.inline.InLineFSUtils;
import org.apache.avro.Schema;
import org.apache.hadoop.conf.Configuration;
@@ -121,7 +120,7 @@ public class HoodieParquetDataBlock extends HoodieDataBlock {
parquetWriter = HoodieFileWriterFactory.getFileWriter(
HoodieFileFormat.PARQUET,
outputStream,
- HadoopFSUtils.getStorageConf(new Configuration()),
+ HoodieStorageUtils.getStorageConf(new Configuration()),
config,
writerSchema,
recordType);
@@ -151,7 +150,7 @@ public class HoodieParquetDataBlock extends HoodieDataBlock {
// NOTE: It's important to extend Hadoop configuration here to make sure configuration
// is appropriately carried over
- StorageConfiguration<?> inlineConf = FSUtils.buildInlineConf(blockContentLoc.getStorageConf());
+ StorageConfiguration<?> inlineConf = blockContentLoc.getStorageConf().getInline();
StoragePath inlineLogFilePath = InLineFSUtils.getInlineFilePath(
blockContentLoc.getLogFile().getPath(),
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/util/BaseFileUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/util/BaseFileUtils.java
index df8325c6476..317a38bfc3e 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/util/BaseFileUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/util/BaseFileUtils.java
@@ -47,12 +47,13 @@ import java.util.Set;
* Utils for Hudi base file.
*/
public abstract class BaseFileUtils {
+ public static final String ORC_UTILS = "org.apache.hudi.common.util.OrcUtils";
public static BaseFileUtils getInstance(String path) {
if (path.endsWith(HoodieFileFormat.PARQUET.getFileExtension())) {
return new ParquetUtils();
} else if (path.endsWith(HoodieFileFormat.ORC.getFileExtension())) {
- return new OrcUtils();
+ return ReflectionUtils.loadClass(ORC_UTILS);
}
throw new UnsupportedOperationException("The format for file " + path + " is not supported yet.");
}
@@ -61,7 +62,7 @@ public abstract class BaseFileUtils {
if (HoodieFileFormat.PARQUET.equals(fileFormat)) {
return new ParquetUtils();
} else if (HoodieFileFormat.ORC.equals(fileFormat)) {
- return new OrcUtils();
+ return ReflectionUtils.loadClass(ORC_UTILS);
}
throw new UnsupportedOperationException(fileFormat.name() + " format not supported yet.");
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/util/ConfigUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/util/ConfigUtils.java
index f528f37437c..643b123d596 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/util/ConfigUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/util/ConfigUtils.java
@@ -20,6 +20,7 @@ package org.apache.hudi.common.util;
import org.apache.hudi.common.config.ConfigProperty;
import org.apache.hudi.common.config.HoodieConfig;
+import org.apache.hudi.common.config.PropertiesConfig;
import org.apache.hudi.common.config.TypedProperties;
import org.apache.hudi.common.model.HoodiePayloadProps;
import org.apache.hudi.common.table.HoodieTableConfig;
@@ -556,4 +557,8 @@ public class ConfigUtils {
Boolean.toString(storageConf.getBoolean(USE_NATIVE_HFILE_READER.key(), USE_NATIVE_HFILE_READER.defaultValue())));
return config;
}
+
+ public static TypedProperties loadGlobalProperties() {
+ return ((PropertiesConfig) ReflectionUtils.loadClass("org.apache.hudi.common.config.DFSPropertiesConfiguration")).getGlobalProperties();
+ }
}
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/util/ParquetUtils.java b/hudi-common/src/main/java/org/apache/hudi/common/util/ParquetUtils.java
index 0bbc203f30d..9298626262d 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/util/ParquetUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/common/util/ParquetUtils.java
@@ -27,9 +27,9 @@ import org.apache.hudi.common.model.HoodieRecord;
import org.apache.hudi.common.util.collection.ClosableIterator;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.exception.MetadataNotFoundException;
-import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.keygen.BaseKeyGenerator;
import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StorageConfiguration;
import org.apache.hudi.storage.StoragePath;
@@ -99,7 +99,8 @@ public class ParquetUtils extends BaseFileUtils {
ParquetMetadata footer;
try {
// TODO(vc): Should we use the parallel reading version here?
- footer = ParquetFileReader.readFooter(HadoopFSUtils.getFs(parquetFileHadoopPath.toString(), conf).getConf(), parquetFileHadoopPath);
+ footer = ParquetFileReader.readFooter(HoodieStorageUtils.getStorage(
+ parquetFileHadoopPath.toString(), conf).getConf().unwrapAs(Configuration.class), parquetFileHadoopPath);
} catch (IOException e) {
throw new HoodieIOException("Failed to read footer for parquet " + parquetFileHadoopPath, e);
}
@@ -123,7 +124,7 @@ public class ParquetUtils extends BaseFileUtils {
filterFunction = Option.of(new RecordKeysFilterFunction(filter));
}
Configuration conf = configuration.unwrapCopyAs(Configuration.class);
- conf.addResource(HadoopFSUtils.getFs(filePath.toString(), conf).getConf());
+ conf.addResource(HoodieStorageUtils.getStorage(filePath.toString(), configuration).getConf().unwrapAs(Configuration.class));
AvroReadSupport.setAvroReadSchema(conf, readSchema);
AvroReadSupport.setRequestedProjection(conf, readSchema);
Set<String> rowKeys = new HashSet<>();
@@ -175,7 +176,7 @@ public class ParquetUtils extends BaseFileUtils {
public ClosableIterator<HoodieKey> getHoodieKeyIterator(StorageConfiguration<?> configuration, StoragePath filePath, Option<BaseKeyGenerator> keyGeneratorOpt) {
try {
Configuration conf = configuration.unwrapCopyAs(Configuration.class);
- conf.addResource(HadoopFSUtils.getFs(filePath.toString(), conf).getConf());
+ conf.addResource(HoodieStorageUtils.getStorage(filePath.toString(), configuration).getConf().unwrapAs(Configuration.class));
Schema readSchema = keyGeneratorOpt
.map(keyGenerator -> {
List<String> fields = new ArrayList<>();
diff --git a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroFileWriterFactory.java b/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroFileWriterFactory.java
index 9b137ce5d9d..2a727158e17 100644
--- a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroFileWriterFactory.java
+++ b/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroFileWriterFactory.java
@@ -26,6 +26,7 @@ import org.apache.hudi.common.engine.TaskContextSupplier;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.ReflectionUtils;
+import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.storage.StorageConfiguration;
import org.apache.hudi.storage.StoragePath;
@@ -47,6 +48,11 @@ import static org.apache.hudi.io.storage.HoodieHFileConfig.HFILE_COMPARATOR;
import static org.apache.hudi.io.storage.HoodieHFileConfig.PREFETCH_ON_OPEN;
public class HoodieAvroFileWriterFactory extends HoodieFileWriterFactory {
+ //hardcoded classes to remove at a later time
+ public static final String HOODIE_AVRO_PARQUET_WRITER = "org.apache.hudi.io.storage.HoodieAvroParquetWriter";
+ public static final String HOODIE_AVRO_HFILE_WRITER = "org.apache.hudi.io.storage.HoodieAvroHFileWriter";
+ public static final String HOODIE_AVRO_ORC_WRITER = "org.apache.hudi.io.storage.HoodieAvroOrcWriter";
+
@Override
protected HoodieFileWriter newParquetFileWriter(
String instantTime, StoragePath path, StorageConfiguration<?> conf, HoodieConfig config, Schema schema,
@@ -66,7 +72,14 @@ public class HoodieAvroFileWriterFactory extends HoodieFileWriterFactory {
config.getLongOrDefault(HoodieStorageConfig.PARQUET_MAX_FILE_SIZE),
conf.unwrapAs(Configuration.class), config.getDoubleOrDefault(HoodieStorageConfig.PARQUET_COMPRESSION_RATIO_FRACTION),
config.getBooleanOrDefault(HoodieStorageConfig.PARQUET_DICTIONARY_ENABLED));
- return new HoodieAvroParquetWriter(path, parquetConfig, instantTime, taskContextSupplier, populateMetaFields);
+ try {
+ return (HoodieFileWriter) ReflectionUtils.loadClass(HOODIE_AVRO_PARQUET_WRITER,
+ new Class<?>[] {StoragePath.class, HoodieParquetConfig.class, String.class, TaskContextSupplier.class, boolean.class},
+ path, parquetConfig, instantTime, taskContextSupplier, populateMetaFields);
+ } catch (HoodieException e) {
+ throw (IOException) e.getCause().getCause();
+ }
+
}
protected HoodieFileWriter newParquetFileWriter(
@@ -94,7 +107,13 @@ public class HoodieAvroFileWriterFactory extends HoodieFileWriterFactory {
HoodieAvroHFileReaderImplBase.KEY_FIELD_NAME,
PREFETCH_ON_OPEN, CACHE_DATA_IN_L1, DROP_BEHIND_CACHE_COMPACTION, filter, HFILE_COMPARATOR);
- return new HoodieAvroHFileWriter(instantTime, path, hfileConfig, schema, taskContextSupplier, config.getBoolean(HoodieTableConfig.POPULATE_META_FIELDS));
+ try {
+ return (HoodieFileWriter) ReflectionUtils.loadClass(HOODIE_AVRO_HFILE_WRITER,
+ new Class<?>[] {String.class, StoragePath.class, HoodieHFileConfig.class, Schema.class, TaskContextSupplier.class, boolean.class},
+ instantTime, path, hfileConfig, schema, taskContextSupplier, config.getBoolean(HoodieTableConfig.POPULATE_META_FIELDS));
+ } catch (HoodieException e) {
+ throw (IOException) e.getCause().getCause();
+ }
}
protected HoodieFileWriter newOrcFileWriter(
@@ -106,7 +125,13 @@ public class HoodieAvroFileWriterFactory extends HoodieFileWriterFactory {
config.getInt(HoodieStorageConfig.ORC_STRIPE_SIZE),
config.getInt(HoodieStorageConfig.ORC_BLOCK_SIZE),
config.getLong(HoodieStorageConfig.ORC_FILE_MAX_SIZE), filter);
- return new HoodieAvroOrcWriter(instantTime, path, orcConfig, schema, taskContextSupplier);
+ try {
+ return (HoodieFileWriter) ReflectionUtils.loadClass(HOODIE_AVRO_ORC_WRITER,
+ new Class<?>[] {String.class, StoragePath.class, HoodieOrcConfig.class, Schema.class, TaskContextSupplier.class},
+ instantTime, path, orcConfig, schema, taskContextSupplier);
+ } catch (HoodieException e) {
+ throw (IOException) e.getCause().getCause();
+ }
}
private HoodieAvroWriteSupport getHoodieAvroWriteSupport(StorageConfiguration<?> conf, Schema schema,
diff --git a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieNativeAvroHFileReader.java b/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieNativeAvroHFileReader.java
index b32e058c78b..e47e393e6a6 100644
--- a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieNativeAvroHFileReader.java
+++ b/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieNativeAvroHFileReader.java
@@ -29,7 +29,6 @@ import org.apache.hudi.common.util.collection.CloseableMappingIterator;
import org.apache.hudi.common.util.io.ByteBufferBackedInputStream;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieIOException;
-import org.apache.hudi.hadoop.fs.HadoopSeekableDataInputStream;
import org.apache.hudi.io.ByteArraySeekableDataInputStream;
import org.apache.hudi.io.SeekableDataInputStream;
import org.apache.hudi.io.hfile.HFileReader;
@@ -45,7 +44,6 @@ import org.apache.hudi.util.Lazy;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.IndexedRecord;
-import org.apache.hadoop.fs.FSDataInputStream;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -261,7 +259,7 @@ public class HoodieNativeAvroHFileReader extends HoodieAvroHFileReaderImplBase {
if (path.isPresent()) {
HoodieStorage storage = HoodieStorageUtils.getStorage(path.get(), conf);
fileSize = storage.getPathInfo(path.get()).getLength();
- inputStream = new HadoopSeekableDataInputStream((FSDataInputStream) storage.open(path.get()));
+ inputStream = storage.openSeekable(path.get(), false);
} else {
fileSize = bytesContent.get().length;
inputStream = new ByteArraySeekableDataInputStream(new ByteBufferBackedInputStream(bytesContent.get()));
diff --git a/hudi-common/src/main/java/org/apache/hudi/metrics/Metrics.java b/hudi-common/src/main/java/org/apache/hudi/metrics/Metrics.java
index 33ae1b75199..af32248eea1 100644
--- a/hudi-common/src/main/java/org/apache/hudi/metrics/Metrics.java
+++ b/hudi-common/src/main/java/org/apache/hudi/metrics/Metrics.java
@@ -23,7 +23,6 @@ import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.config.metrics.HoodieMetricsConfig;
-import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StoragePath;
@@ -99,8 +98,7 @@ public class Metrics {
private List<MetricsReporter> addAdditionalMetricsExporters(HoodieMetricsConfig metricConfig) {
List<MetricsReporter> reporterList = new ArrayList<>();
List<String> propPathList = StringUtils.split(metricConfig.getMetricReporterFileBasedConfigs(), ",");
- try (HoodieStorage storage = HoodieStorageUtils.getStorage(
- propPathList.get(0), HadoopFSUtils.getStorageConf(new Configuration()))) {
+ try (HoodieStorage storage = HoodieStorageUtils.getStorage(propPathList.get(0), new Configuration())) {
for (String propPath : propPathList) {
HoodieMetricsConfig secondarySourceConfig = HoodieMetricsConfig.newBuilder().fromInputStream(
storage.open(new StoragePath(propPath))).withPath(metricConfig.getBasePath()).build();
diff --git a/hudi-common/src/main/java/org/apache/hudi/storage/HoodieStorageUtils.java b/hudi-common/src/main/java/org/apache/hudi/storage/HoodieStorageUtils.java
index da6efc3e925..64bcde90d71 100644
--- a/hudi-common/src/main/java/org/apache/hudi/storage/HoodieStorageUtils.java
+++ b/hudi-common/src/main/java/org/apache/hudi/storage/HoodieStorageUtils.java
@@ -19,14 +19,15 @@
package org.apache.hudi.storage;
-import org.apache.hudi.hadoop.fs.HadoopFSUtils;
-import org.apache.hudi.hadoop.fs.HoodieWrapperFileSystem;
-import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
+import org.apache.hudi.common.fs.ConsistencyGuard;
+import org.apache.hudi.common.util.ReflectionUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
public class HoodieStorageUtils {
+ public static final String HUDI_HADOOP_STORAGE = "org.apache.hudi.storage.hadoop.HoodieHadoopStorage";
+ public static final String HADOOP_STORAGE_CONF = "org.apache.hudi.storage.hadoop.HadoopStorageConfiguration";
public static final String DEFAULT_URI = "file:///";
public static HoodieStorage getStorage(StorageConfiguration<?> conf) {
@@ -34,22 +35,47 @@ public class HoodieStorageUtils {
}
public static HoodieStorage getStorage(FileSystem fs) {
- return new HoodieHadoopStorage(fs);
+ return (HoodieStorage) ReflectionUtils.loadClass(HUDI_HADOOP_STORAGE, new Class<?>[] {FileSystem.class}, fs);
}
public static HoodieStorage getStorage(String basePath, StorageConfiguration<?> conf) {
- return getStorage(HadoopFSUtils.getFs(basePath, conf));
+ return (HoodieStorage) ReflectionUtils.loadClass(HUDI_HADOOP_STORAGE, new Class<?>[] {String.class, StorageConfiguration.class}, basePath, conf);
+ }
+
+ public static HoodieStorage getStorage(String basePath, Configuration conf) {
+ return (HoodieStorage) ReflectionUtils.loadClass(HUDI_HADOOP_STORAGE, new Class<?>[] {String.class, Configuration.class}, basePath, conf);
}
public static HoodieStorage getStorage(StoragePath path, StorageConfiguration<?> conf) {
- return getStorage(HadoopFSUtils.getFs(path, conf.unwrapAs(Configuration.class)));
+ return (HoodieStorage) ReflectionUtils.loadClass(HUDI_HADOOP_STORAGE, new Class<?>[] {StoragePath.class, StorageConfiguration.class}, path, conf);
+ }
+
+ public static HoodieStorage getStorage(StoragePath path,
+ StorageConfiguration<?> conf,
+ boolean enableRetry,
+ long maxRetryIntervalMs,
+ int maxRetryNumbers,
+ long initialRetryIntervalMs,
+ String retryExceptions,
+ ConsistencyGuard consistencyGuard) {
+ return (HoodieStorage) ReflectionUtils.loadClass(HUDI_HADOOP_STORAGE,
+ new Class<?>[] {StoragePath.class, StorageConfiguration.class, boolean.class, long.class, int.class, long.class,
+ String.class, ConsistencyGuard.class},
+ path, conf, enableRetry, maxRetryIntervalMs, maxRetryNumbers, initialRetryIntervalMs, retryExceptions,
+ consistencyGuard);
}
public static HoodieStorage getRawStorage(HoodieStorage storage) {
- FileSystem fs = (FileSystem) storage.getFileSystem();
- if (fs instanceof HoodieWrapperFileSystem) {
- return getStorage(((HoodieWrapperFileSystem) fs).getFileSystem());
- }
- return storage;
+ return (HoodieStorage) ReflectionUtils.loadClass(HUDI_HADOOP_STORAGE, new Class<?>[] {HoodieStorage.class}, storage);
+ }
+
+ public static StorageConfiguration<?> getStorageConf(Configuration conf) {
+ return (StorageConfiguration<?>) ReflectionUtils.loadClass(HADOOP_STORAGE_CONF,
+ new Class<?>[] {Configuration.class}, conf);
+ }
+
+ public static StorageConfiguration<Configuration> getStorageConfWithCopy(Configuration conf) {
+ return (StorageConfiguration<Configuration>) ReflectionUtils.loadClass(HADOOP_STORAGE_CONF,
+ new Class<?>[] {Configuration.class, boolean.class}, conf, true);
}
}
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsMocked.java b/hudi-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsMocked.java
deleted file mode 100644
index dba2da30672..00000000000
--- a/hudi-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsMocked.java
+++ /dev/null
@@ -1,120 +0,0 @@
-/*
- * 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.hudi.common.fs;
-
-import org.apache.hudi.common.util.Option;
-import org.apache.hudi.exception.HoodieIOException;
-import org.apache.hudi.storage.HoodieStorage;
-import org.apache.hudi.storage.StoragePath;
-import org.apache.hudi.storage.StoragePathInfo;
-
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import org.mockito.Mock;
-import org.mockito.MockitoAnnotations;
-
-import java.io.FileNotFoundException;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.eq;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
-
-public class TestFSUtilsMocked {
-
- @Mock
- private HoodieStorage mockStorage;
-
- private final StoragePath basePath = new StoragePath("/base/path");
- private final Set<String> fileNames = new HashSet<>(Arrays.asList("file1.txt", "file2.txt"));
- private StoragePathInfo mockFile1;
- private StoragePathInfo mockFile2;
-
- @BeforeEach
- public void setUp() {
- MockitoAnnotations.initMocks(this);
- mockFile1 = new StoragePathInfo(new StoragePath("/base/path/file1.txt"), 100, false, (short) 3, 1024, 0);
- mockFile2 = new StoragePathInfo(new StoragePath("/base/path/file2.txt"), 200, false, (short) 3, 1024, 0);
- }
-
- @Test
- public void testGetPathInfoUnderPartitionWithListStatus() throws IOException, IOException {
- // Setup
- when(mockStorage.getScheme()).thenReturn("file"); // Assuming "file" is list status friendly
- List<StoragePathInfo> listingResult = new ArrayList<>();
- listingResult.add(mockFile1);
- listingResult.add(mockFile2);
- when(mockStorage.listDirectEntries(eq(basePath), any())).thenReturn(listingResult);
-
- // Execute
- List<Option<StoragePathInfo>> result = FSUtils.getPathInfoUnderPartition(mockStorage, basePath, fileNames, false);
-
- // Verify
- assertEquals(2, result.size());
- assertTrue(result.get(0).isPresent());
- assertTrue(result.get(1).isPresent());
-
- // Cleanup
- verify(mockStorage, times(1)).listDirectEntries((StoragePath) any(), any());
- }
-
- @Test
- public void testGetPathInfoUnderPartitionIgnoringMissingFiles() throws IOException {
- // Setup for scenario where file2.txt does not exist
- when(mockStorage.getScheme()).thenReturn("hdfs"); // Assuming "hdfs" is not list status friendly
- when(mockStorage.getPathInfo(new StoragePath("/base/path/file1.txt"))).thenReturn(mockFile1);
- when(mockStorage.getPathInfo(new StoragePath("/base/path/file2.txt"))).thenThrow(new FileNotFoundException());
-
- // Execute
- List<Option<StoragePathInfo>> result = FSUtils.getPathInfoUnderPartition(mockStorage, basePath, fileNames, true);
-
- // Verify
- assertEquals(2, result.size());
- assertTrue(result.get(0).isPresent());
- assertFalse(result.get(1).isPresent()); // Missing file results in an empty Option
-
- // Cleanup
- verify(mockStorage, times(2)).getPathInfo(any());
- }
-
- @Test
- public void testGetPathInfoUnderPartitionThrowsHoodieIOException() throws IOException {
- // Setup
- when(mockStorage.getScheme()).thenReturn("file"); // Assuming "file" is list status friendly
- when(mockStorage.listDirectEntries((StoragePath) any(), any())).thenThrow(new IOException());
-
- // Execute & Verify
- assertThrows(HoodieIOException.class, () ->
- FSUtils.getPathInfoUnderPartition(mockStorage, basePath, fileNames, false));
-
- // Cleanup
- verify(mockStorage, times(1)).listDirectEntries((StoragePath) any(), any());
- }
-}
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java b/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java
index 7c9e111f59e..617986be286 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestDataGenerator.java
@@ -37,8 +37,8 @@ import org.apache.hudi.common.util.AvroOrcUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieIOException;
-import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.storage.HoodieStorage;
+import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StorageConfiguration;
import org.apache.hudi.storage.StoragePath;
@@ -49,8 +49,6 @@ import org.apache.avro.generic.GenericArray;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericFixed;
import org.apache.avro.generic.GenericRecord;
-import org.apache.hadoop.fs.FSDataOutputStream;
-import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.orc.TypeDescription;
import org.slf4j.Logger;
@@ -557,8 +555,8 @@ public class HoodieTestDataGenerator implements AutoCloseable {
basePath + "/" + HoodieTableMetaClient.METAFOLDER_NAME + "/" + f);
OutputStream os = null;
try {
- FileSystem fs = HadoopFSUtils.getFs(basePath, configuration);
- os = fs.create(commitFile, true);
+ HoodieStorage storage = HoodieStorageUtils.getStorage(basePath, configuration);
+ os = storage.create(new StoragePath(commitFile.toUri()), true);
// Write empty commit metadata
os.write(content);
} catch (IOException ioe) {
@@ -607,8 +605,8 @@ public class HoodieTestDataGenerator implements AutoCloseable {
}
private static void createEmptyFile(String basePath, Path filePath, StorageConfiguration<?> configuration) throws IOException {
- FileSystem fs = HadoopFSUtils.getFs(basePath, configuration);
- OutputStream os = fs.create(filePath, true);
+ HoodieStorage storage = HoodieStorageUtils.getStorage(basePath, configuration);
+ OutputStream os = storage.create(new StoragePath(filePath.toUri()), true);
os.close();
}
@@ -623,8 +621,8 @@ public class HoodieTestDataGenerator implements AutoCloseable {
StorageConfiguration<?> configuration) throws IOException {
Path commitFile =
new Path(basePath + "/" + HoodieTableMetaClient.AUXILIARYFOLDER_NAME + "/" + instant.getFileName());
- FileSystem fs = HadoopFSUtils.getFs(basePath, configuration);
- try (OutputStream os = fs.create(commitFile, true)) {
+ HoodieStorage storage = HoodieStorageUtils.getStorage(basePath, configuration);
+ try (OutputStream os = storage.create(new StoragePath(commitFile.toUri()), true)) {
HoodieCompactionPlan workload = HoodieCompactionPlan.newBuilder().setVersion(1).build();
// Write empty commit metadata
os.write(TimelineMetadataUtils.serializeCompactionPlan(workload).get());
@@ -633,13 +631,13 @@ public class HoodieTestDataGenerator implements AutoCloseable {
public static void createSavepointFile(String basePath, String instantTime, StorageConfiguration<?> configuration)
throws IOException {
- Path commitFile = new Path(basePath + "/" + HoodieTableMetaClient.METAFOLDER_NAME + "/"
+ StoragePath commitFile = new StoragePath(basePath + "/" + HoodieTableMetaClient.METAFOLDER_NAME + "/"
+ HoodieTimeline.makeSavePointFileName(instantTime));
- FileSystem fs = HadoopFSUtils.getFs(basePath, configuration);
- try (FSDataOutputStream os = fs.create(commitFile, true)) {
+ HoodieStorage storage = HoodieStorageUtils.getStorage(basePath, configuration);
+ try (OutputStream os = storage.create(commitFile, true)) {
HoodieCommitMetadata commitMetadata = new HoodieCommitMetadata();
// Write empty commit metadata
- os.writeBytes(new String(getUTF8Bytes(commitMetadata.toJsonString())));
+ os.write(getUTF8Bytes(commitMetadata.toJsonString()));
}
}
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestUtils.java b/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestUtils.java
index e61f8f4c632..d0af0ae8963 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestUtils.java
+++ b/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestUtils.java
@@ -26,7 +26,7 @@ import org.apache.hudi.common.model.HoodieWriteStat;
import org.apache.hudi.common.model.HoodieWriteStat.RuntimeStats;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
-import org.apache.hudi.hadoop.fs.HadoopFSUtils;
+import org.apache.hudi.common.util.ReflectionUtils;
import org.apache.hudi.metadata.HoodieTableMetadata;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.HoodieStorageUtils;
@@ -51,6 +51,8 @@ import java.util.Objects;
import java.util.Properties;
import java.util.UUID;
+import static org.apache.hudi.storage.HoodieStorageUtils.HADOOP_STORAGE_CONF;
+
/**
* A utility class for testing.
*/
@@ -63,7 +65,13 @@ public class HoodieTestUtils {
public static final String[] DEFAULT_PARTITION_PATHS = {"2016/03/15", "2015/03/16", "2015/03/17"};
public static StorageConfiguration<Configuration> getDefaultStorageConf() {
- return HadoopFSUtils.getStorageConf(new Configuration(false));
+ return (StorageConfiguration<Configuration>) ReflectionUtils.loadClass(HADOOP_STORAGE_CONF,
+ new Class<?>[] {Boolean.class}, false);
+ }
+
+ public static StorageConfiguration<Configuration> getDefaultStorageConfWithDefaults() {
+ return (StorageConfiguration<Configuration>) ReflectionUtils.loadClass(HADOOP_STORAGE_CONF,
+ new Class<?>[] {Boolean.class}, true);
}
public static HoodieStorage getStorage(String path) {
@@ -210,7 +218,7 @@ public class HoodieTestUtils {
*/
public static HoodieTableMetaClient createMetaClient(Configuration conf,
String basePath) {
- return createMetaClient(HadoopFSUtils.getStorageConfWithCopy(conf), basePath);
+ return createMetaClient(HoodieStorageUtils.getStorageConfWithCopy(conf), basePath);
}
/**
diff --git a/hudi-examples/hudi-examples-common/pom.xml b/hudi-examples/hudi-examples-common/pom.xml
index ff627329fe3..1fcfaec34fd 100644
--- a/hudi-examples/hudi-examples-common/pom.xml
+++ b/hudi-examples/hudi-examples-common/pom.xml
@@ -94,6 +94,11 @@
<artifactId>hudi-common</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<!-- Kryo -->
<dependency>
diff --git a/hudi-examples/hudi-examples-flink/pom.xml b/hudi-examples/hudi-examples-flink/pom.xml
index 82e519b9ac5..4d5e305d948 100644
--- a/hudi-examples/hudi-examples-flink/pom.xml
+++ b/hudi-examples/hudi-examples-flink/pom.xml
@@ -79,6 +79,11 @@
<artifactId>hudi-common</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-client-common</artifactId>
@@ -272,6 +277,14 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-client-common</artifactId>
diff --git a/hudi-flink-datasource/hudi-flink/pom.xml b/hudi-flink-datasource/hudi-flink/pom.xml
index 9cdcfb426e1..827494e74fd 100644
--- a/hudi-flink-datasource/hudi-flink/pom.xml
+++ b/hudi-flink-datasource/hudi-flink/pom.xml
@@ -95,6 +95,11 @@
<artifactId>hudi-common</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-client-common</artifactId>
@@ -349,6 +354,14 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-client-common</artifactId>
diff --git a/hudi-hadoop-common/pom.xml b/hudi-hadoop-common/pom.xml
index e4fbf2d94a9..9e7f7bc8c3f 100644
--- a/hudi-hadoop-common/pom.xml
+++ b/hudi-hadoop-common/pom.xml
@@ -68,6 +68,11 @@
</build>
<dependencies>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-io</artifactId>
@@ -92,6 +97,17 @@
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.parquet</groupId>
+ <artifactId>parquet-avro</artifactId>
+ </dependency>
+
+ <!-- Kryo -->
+ <dependency>
+ <groupId>com.esotericsoftware</groupId>
+ <artifactId>kryo-shaded</artifactId>
+ </dependency>
+
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-tests-common</artifactId>
@@ -106,5 +122,19 @@
<version>${project.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.github.stefanbirkner</groupId>
+ <artifactId>system-rules</artifactId>
+ <version>1.17.2</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/config/DFSPropertiesConfiguration.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/common/config/DFSPropertiesConfiguration.java
similarity index 98%
rename from hudi-common/src/main/java/org/apache/hudi/common/config/DFSPropertiesConfiguration.java
rename to hudi-hadoop-common/src/main/java/org/apache/hudi/common/config/DFSPropertiesConfiguration.java
index f7987b870d1..cc706dfd719 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/config/DFSPropertiesConfiguration.java
+++ b/hudi-hadoop-common/src/main/java/org/apache/hudi/common/config/DFSPropertiesConfiguration.java
@@ -55,7 +55,7 @@ import java.util.Set;
*
* Note: Not reusing commons-configuration since it has too many conflicting runtime deps.
*/
-public class DFSPropertiesConfiguration {
+public class DFSPropertiesConfiguration extends PropertiesConfig {
private static final Logger LOG = LoggerFactory.getLogger(DFSPropertiesConfiguration.class);
@@ -202,6 +202,11 @@ public class DFSPropertiesConfiguration {
}
}
+ @Override
+ public TypedProperties getGlobalProperties() {
+ return getGlobalProps();
+ }
+
public static TypedProperties getGlobalProps() {
final TypedProperties globalProps = new TypedProperties();
globalProps.putAll(GLOBAL_PROPS);
diff --git a/hudi-common/src/main/java/org/apache/hudi/common/util/OrcUtils.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/common/util/OrcUtils.java
similarity index 93%
rename from hudi-common/src/main/java/org/apache/hudi/common/util/OrcUtils.java
rename to hudi-hadoop-common/src/main/java/org/apache/hudi/common/util/OrcUtils.java
index 9cab5d58877..d0f51763e8d 100644
--- a/hudi-common/src/main/java/org/apache/hudi/common/util/OrcUtils.java
+++ b/hudi-hadoop-common/src/main/java/org/apache/hudi/common/util/OrcUtils.java
@@ -62,6 +62,7 @@ import java.util.stream.Collectors;
import static org.apache.hudi.common.util.BinaryUtil.toBytes;
import static org.apache.hudi.common.util.StringUtils.getUTF8Bytes;
+import static org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToHadoopPath;
/**
* Utility functions for ORC files.
@@ -80,7 +81,7 @@ public class OrcUtils extends BaseFileUtils {
try {
Configuration conf = configuration.unwrapCopyAs(Configuration.class);
conf.addResource(HadoopFSUtils.getFs(filePath.toString(), conf).getConf());
- Reader reader = OrcFile.createReader(new Path(filePath.toUri()), OrcFile.readerOptions(conf));
+ Reader reader = OrcFile.createReader(convertToHadoopPath(filePath), OrcFile.readerOptions(conf));
Schema readSchema = HoodieAvroUtils.getRecordKeyPartitionPathSchema();
TypeDescription orcSchema = AvroOrcUtils.createOrcSchema(readSchema);
@@ -147,7 +148,7 @@ public class OrcUtils extends BaseFileUtils {
public List<GenericRecord> readAvroRecords(StorageConfiguration<?> configuration, StoragePath filePath) {
Schema avroSchema;
try (Reader reader = OrcFile.createReader(
- new Path(filePath.toUri()), OrcFile.readerOptions(configuration.unwrapAs(Configuration.class)))) {
+ convertToHadoopPath(filePath), OrcFile.readerOptions(configuration.unwrapAs(Configuration.class)))) {
avroSchema = AvroOrcUtils.createAvroSchema(reader.getSchema());
} catch (IOException io) {
throw new HoodieIOException("Unable to read Avro records from an ORC file:" + filePath, io);
@@ -162,7 +163,7 @@ public class OrcUtils extends BaseFileUtils {
public List<GenericRecord> readAvroRecords(StorageConfiguration<?> configuration, StoragePath filePath, Schema avroSchema) {
List<GenericRecord> records = new ArrayList<>();
try (Reader reader = OrcFile.createReader(
- new Path(filePath.toUri()), OrcFile.readerOptions(configuration.unwrapAs(Configuration.class)))) {
+ convertToHadoopPath(filePath), OrcFile.readerOptions(configuration.unwrapAs(Configuration.class)))) {
TypeDescription orcSchema = reader.getSchema();
try (RecordReader recordReader = reader.rows(
new Options(configuration.unwrapAs(Configuration.class)).schema(orcSchema))) {
@@ -228,7 +229,7 @@ public class OrcUtils extends BaseFileUtils {
public Map<String, String> readFooter(StorageConfiguration<?> conf, boolean required,
StoragePath filePath, String... footerNames) {
try (Reader reader = OrcFile.createReader(
- new Path(filePath.toUri()), OrcFile.readerOptions(conf.unwrapAs(Configuration.class)))) {
+ convertToHadoopPath(filePath), OrcFile.readerOptions(conf.unwrapAs(Configuration.class)))) {
Map<String, String> footerVals = new HashMap<>();
List<UserMetadataItem> metadataItemList = reader.getFileTail().getFooter().getMetadataList();
Map<String, String> metadata = metadataItemList.stream().collect(Collectors.toMap(
@@ -251,7 +252,7 @@ public class OrcUtils extends BaseFileUtils {
@Override
public Schema readAvroSchema(StorageConfiguration<?> conf, StoragePath filePath) {
try (Reader reader = OrcFile.createReader(
- new Path(filePath.toUri()), OrcFile.readerOptions(conf.unwrapAs(Configuration.class)))) {
+ convertToHadoopPath(filePath), OrcFile.readerOptions(conf.unwrapAs(Configuration.class)))) {
if (reader.hasMetadataValue("orc.avro.schema")) {
ByteBuffer metadataValue = reader.getMetadataValue("orc.avro.schema");
byte[] bytes = toBytes(metadataValue);
@@ -273,7 +274,7 @@ public class OrcUtils extends BaseFileUtils {
@Override
public long getRowCount(StorageConfiguration<?> conf, StoragePath filePath) {
try (Reader reader = OrcFile.createReader(
- new Path(filePath.toUri()), OrcFile.readerOptions(conf.unwrapAs(Configuration.class)))) {
+ convertToHadoopPath(filePath), OrcFile.readerOptions(conf.unwrapAs(Configuration.class)))) {
return reader.getNumberOfRows();
} catch (IOException io) {
throw new HoodieIOException("Unable to get row count for ORC file:" + filePath, io);
@@ -288,7 +289,7 @@ public class OrcUtils extends BaseFileUtils {
OrcFile.WriterOptions writerOptions = OrcFile.writerOptions(storage.unwrapConfAs(Configuration.class))
.fileSystem((FileSystem) storage.getFileSystem())
.setSchema(AvroOrcUtils.createOrcSchema(schema));
- try (Writer writer = OrcFile.createWriter(new Path(filePath.toUri()), writerOptions)) {
+ try (Writer writer = OrcFile.createWriter(convertToHadoopPath(filePath), writerOptions)) {
for (String key : props.stringPropertyNames()) {
writer.addUserMetadata(key, ByteBuffer.wrap(getUTF8Bytes(props.getProperty(key))));
}
diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HadoopFSUtils.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HadoopFSUtils.java
index 3aa66e6c2de..3119ee8c0c0 100644
--- a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HadoopFSUtils.java
+++ b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HadoopFSUtils.java
@@ -20,13 +20,11 @@
package org.apache.hudi.hadoop.fs;
import org.apache.hudi.exception.HoodieIOException;
-import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StorageConfiguration;
import org.apache.hudi.storage.StoragePath;
import org.apache.hudi.storage.StoragePathInfo;
import org.apache.hudi.storage.StorageSchemes;
import org.apache.hudi.storage.hadoop.HadoopStorageConfiguration;
-import org.apache.hudi.storage.hadoop.HoodieHadoopStorage;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BufferedFSInputStream;
@@ -42,8 +40,6 @@ import java.io.File;
import java.io.IOException;
import java.util.Map;
-import static org.apache.hudi.common.util.ValidationUtils.checkArgument;
-
/**
* Utility functions related to accessing the file storage on Hadoop.
*/
@@ -88,7 +84,7 @@ public class HadoopFSUtils {
}
public static FileSystem getFs(StoragePath path, Configuration conf) {
- return getFs(new Path(path.toUri()), conf);
+ return getFs(convertToHadoopPath(path), conf);
}
public static FileSystem getFs(Path path, Configuration conf) {
@@ -109,25 +105,6 @@ public class HadoopFSUtils {
return getFs(pathStr, conf);
}
- public static HoodieStorage getStorageWithWrapperFS(StoragePath path,
- StorageConfiguration<?> conf,
- boolean enableRetry,
- long maxRetryIntervalMs,
- int maxRetryNumbers,
- long initialRetryIntervalMs,
- String retryExceptions,
- ConsistencyGuard consistencyGuard) {
- FileSystem fileSystem = getFs(path, conf.unwrapCopyAs(Configuration.class));
-
- if (enableRetry) {
- fileSystem = new HoodieRetryWrapperFileSystem(fileSystem,
- maxRetryIntervalMs, maxRetryNumbers, initialRetryIntervalMs, retryExceptions);
- }
- checkArgument(!(fileSystem instanceof HoodieWrapperFileSystem),
- "File System not expected to be that of HoodieWrapperFileSystem");
- return new HoodieHadoopStorage(new HoodieWrapperFileSystem(fileSystem, consistencyGuard));
- }
-
public static Path addSchemeIfLocalPath(String path) {
Path providedPath = new Path(path);
File localFile = new File(path);
@@ -190,11 +167,13 @@ public class HadoopFSUtils {
* @param fs instance of {@link FileSystem} in use.
* @param filePath path of the file.
* @param bufferSize buffer size to be used.
+ * @param wrapStream if false, don't attempt to wrap the stream
* @return the right {@link FSDataInputStream} as required.
*/
public static FSDataInputStream getFSDataInputStream(FileSystem fs,
StoragePath filePath,
- int bufferSize) {
+ int bufferSize,
+ boolean wrapStream) {
FSDataInputStream fsDataInputStream = null;
try {
fsDataInputStream = fs.open(convertToHadoopPath(filePath), bufferSize);
@@ -202,6 +181,10 @@ public class HadoopFSUtils {
throw new HoodieIOException(String.format("Exception creating input stream from file: %s", filePath), e);
}
+ if (!wrapStream) {
+ return fsDataInputStream;
+ }
+
if (isGCSFileSystem(fs)) {
// in GCS FS, we might need to interceptor seek offsets as we might get EOF exception
return new SchemeAwareFSDataInputStream(getFSDataInputStreamForGCS(fsDataInputStream, filePath, bufferSize), true);
@@ -273,4 +256,12 @@ public class HadoopFSUtils {
private static StorageConfiguration<Configuration> getStorageConf(Configuration conf, boolean copy) {
return new HadoopStorageConfiguration(conf, copy);
}
+
+ public static Configuration registerFileSystem(StoragePath file, Configuration conf) {
+ Configuration returnConf = new Configuration(conf);
+ String scheme = HadoopFSUtils.getFs(file.toString(), conf).getScheme();
+ returnConf.set("fs." + HoodieWrapperFileSystem.getHoodieScheme(scheme) + ".impl",
+ HoodieWrapperFileSystem.class.getName());
+ return returnConf;
+ }
}
diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HoodieWrapperFileSystem.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HoodieWrapperFileSystem.java
index 927849fea79..b2a3a97d3bd 100644
--- a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HoodieWrapperFileSystem.java
+++ b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/HoodieWrapperFileSystem.java
@@ -19,6 +19,8 @@
package org.apache.hudi.hadoop.fs;
+import org.apache.hudi.common.fs.ConsistencyGuard;
+import org.apache.hudi.common.fs.NoOpConsistencyGuard;
import org.apache.hudi.common.metrics.Registry;
import org.apache.hudi.common.util.HoodieTimer;
import org.apache.hudi.exception.HoodieException;
@@ -61,6 +63,9 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeoutException;
+import static org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToHadoopPath;
+import static org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToStoragePath;
+
/**
* HoodieWrapperFileSystem wraps the default file system. It holds state about the open streams in the file system to
* support getting the written size to each of the open streams.
@@ -142,7 +147,7 @@ public class HoodieWrapperFileSystem extends FileSystem {
public static Path convertToHoodiePath(StoragePath file, Configuration conf) {
try {
String scheme = HadoopFSUtils.getFs(file.toString(), conf).getScheme();
- return convertPathWithScheme(new Path(file.toUri()), getHoodieScheme(scheme));
+ return convertPathWithScheme(convertToHadoopPath(file), getHoodieScheme(scheme));
} catch (HoodieIOException e) {
throw e;
}
@@ -357,7 +362,7 @@ public class HoodieWrapperFileSystem extends FileSystem {
if (success) {
try {
- consistencyGuard.waitTillFileDisappears(new StoragePath(f.toUri()));
+ consistencyGuard.waitTillFileDisappears(convertToStoragePath(f));
} catch (TimeoutException e) {
throw new HoodieException("Timed out waiting for " + f + " to disappear", e);
}
@@ -969,7 +974,7 @@ public class HoodieWrapperFileSystem extends FileSystem {
}
private StoragePath convertToDefaultStoragePath(Path oldPath) {
- return new StoragePath(convertPathWithScheme(oldPath, getScheme()).toUri());
+ return convertToStoragePath(convertPathWithScheme(oldPath, getScheme()));
}
private Path convertToLocalPath(Path oldPath) {
diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/SizeAwareFSDataOutputStream.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/SizeAwareFSDataOutputStream.java
index 3665c2a69a2..e2851a35084 100644
--- a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/SizeAwareFSDataOutputStream.java
+++ b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/SizeAwareFSDataOutputStream.java
@@ -19,8 +19,8 @@
package org.apache.hudi.hadoop.fs;
+import org.apache.hudi.common.fs.ConsistencyGuard;
import org.apache.hudi.exception.HoodieException;
-import org.apache.hudi.storage.StoragePath;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
@@ -29,6 +29,8 @@ import java.io.IOException;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;
+import static org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToStoragePath;
+
/**
* Wrapper over <code>FSDataOutputStream</code> to keep track of the size of the written bytes. This gives a cheap way
* to check on the underlying file size.
@@ -76,7 +78,7 @@ public class SizeAwareFSDataOutputStream extends FSDataOutputStream {
public void close() throws IOException {
super.close();
try {
- consistencyGuard.waitTillFileAppears(new StoragePath(path.toUri()));
+ consistencyGuard.waitTillFileAppears(convertToStoragePath(path));
} catch (TimeoutException e) {
throw new HoodieException(e);
}
diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/inline/HadoopInLineFSUtils.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/inline/HadoopInLineFSUtils.java
new file mode 100644
index 00000000000..ba252ef45ba
--- /dev/null
+++ b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/inline/HadoopInLineFSUtils.java
@@ -0,0 +1,66 @@
+/*
+ * 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.hudi.hadoop.fs.inline;
+
+import org.apache.hudi.storage.StorageConfiguration;
+import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.inline.InLineFSUtils;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+
+import static org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToHadoopPath;
+import static org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToStoragePath;
+
+/**
+ * Utils to parse InLineFileSystem paths.
+ * Inline FS format:
+ * "inlinefs://<path_to_outer_file>/<outer_file_scheme>/?start_offset=start_offset>&length=<length>"
+ * Eg: "inlinefs://<path_to_outer_file>/s3a/?start_offset=20&length=40"
+ */
+public class HadoopInLineFSUtils extends InLineFSUtils {
+
+ public static StorageConfiguration<Configuration> buildInlineConf(StorageConfiguration<Configuration> storageConf) {
+ StorageConfiguration<Configuration> inlineConf = storageConf.newInstance();
+ inlineConf.set("fs." + InLineFileSystem.SCHEME + ".impl", InLineFileSystem.class.getName());
+ (inlineConf.unwrapAs(Configuration.class)).setClassLoader(InLineFileSystem.class.getClassLoader());
+ return inlineConf;
+ }
+
+ /**
+ * InlineFS Path format:
+ * "inlinefs://path/to/outer/file/outer_file_scheme/?start_offset=start_offset>&length=<length>"
+ * <p>
+ * Outer File Path format:
+ * "outer_file_scheme://path/to/outer/file"
+ * <p>
+ * Example
+ * Input: "inlinefs://file1/s3a/?start_offset=20&length=40".
+ * Output: "s3a://file1"
+ *
+ * @param inlineFSPath InLineFS Path to get the outer file Path
+ * @return Outer file Path from the InLineFS Path
+ */
+ public static Path getOuterFilePathFromInlinePath(Path inlineFSPath) {
+ StoragePath storagePath = convertToStoragePath(inlineFSPath);
+ StoragePath outerFilePath = getOuterFilePathFromInlinePath(storagePath);
+ return convertToHadoopPath(outerFilePath);
+ }
+}
diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/inline/InLineFileSystem.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/inline/InLineFileSystem.java
index 9d7d187b807..9296b717899 100644
--- a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/inline/InLineFileSystem.java
+++ b/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/inline/InLineFileSystem.java
@@ -20,6 +20,7 @@
package org.apache.hudi.hadoop.fs.inline;
import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.inline.InLineFSUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
@@ -33,6 +34,8 @@ import org.apache.hadoop.util.Progressable;
import java.io.IOException;
import java.net.URI;
+import static org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToStoragePath;
+
/**
* Enables reading any inline file at a given offset and length. This {@link FileSystem} is used only in read path and does not support
* any write apis.
@@ -46,7 +49,7 @@ import java.net.URI;
*/
public class InLineFileSystem extends FileSystem {
- public static final String SCHEME = "inlinefs";
+ public static final String SCHEME = InLineFSUtils.SCHEME;
private Configuration conf = null;
@Override
@@ -67,11 +70,11 @@ public class InLineFileSystem extends FileSystem {
@Override
public FSDataInputStream open(Path inlinePath, int bufferSize) throws IOException {
- Path outerPath = InLineFSUtils.getOuterFilePathFromInlinePath(inlinePath);
+ Path outerPath = HadoopInLineFSUtils.getOuterFilePathFromInlinePath(inlinePath);
FileSystem outerFs = outerPath.getFileSystem(conf);
FSDataInputStream outerStream = outerFs.open(outerPath, bufferSize);
- StoragePath inlineStoragePath = new StoragePath(inlinePath.toUri());
- return new InLineFsDataInputStream(InLineFSUtils.startOffset(inlineStoragePath), outerStream, InLineFSUtils.length(inlineStoragePath));
+ StoragePath inlineStoragePath = convertToStoragePath(inlinePath);
+ return new InLineFsDataInputStream(HadoopInLineFSUtils.startOffset(inlineStoragePath), outerStream, HadoopInLineFSUtils.length(inlineStoragePath));
}
@Override
@@ -85,10 +88,10 @@ public class InLineFileSystem extends FileSystem {
@Override
public FileStatus getFileStatus(Path inlinePath) throws IOException {
- Path outerPath = InLineFSUtils.getOuterFilePathFromInlinePath(inlinePath);
+ Path outerPath = HadoopInLineFSUtils.getOuterFilePathFromInlinePath(inlinePath);
FileSystem outerFs = outerPath.getFileSystem(conf);
FileStatus status = outerFs.getFileStatus(outerPath);
- FileStatus toReturn = new FileStatus(InLineFSUtils.length(new StoragePath(inlinePath.toUri())), status.isDirectory(), status.getReplication(), status.getBlockSize(),
+ FileStatus toReturn = new FileStatus(HadoopInLineFSUtils.length(convertToStoragePath(inlinePath)), status.isDirectory(), status.getReplication(), status.getBlockSize(),
status.getModificationTime(), status.getAccessTime(), status.getPermission(), status.getOwner(),
status.getGroup(), inlinePath);
return toReturn;
diff --git a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroHFileWriter.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroHFileWriter.java
similarity index 98%
rename from hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroHFileWriter.java
rename to hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroHFileWriter.java
index 8582144e2f6..a1ffef280f5 100644
--- a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroHFileWriter.java
+++ b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroHFileWriter.java
@@ -21,10 +21,10 @@ package org.apache.hudi.io.storage;
import org.apache.hudi.avro.HoodieAvroUtils;
import org.apache.hudi.common.bloom.BloomFilter;
import org.apache.hudi.common.engine.TaskContextSupplier;
-import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.exception.HoodieDuplicateKeyException;
+import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.hadoop.fs.HoodieWrapperFileSystem;
import org.apache.hudi.storage.StoragePath;
@@ -83,7 +83,7 @@ public class HoodieAvroHFileWriter
public HoodieAvroHFileWriter(String instantTime, StoragePath file, HoodieHFileConfig hfileConfig, Schema schema,
TaskContextSupplier taskContextSupplier, boolean populateMetaFields) throws IOException {
- Configuration conf = FSUtils.registerFileSystem(file, hfileConfig.getHadoopConf());
+ Configuration conf = HadoopFSUtils.registerFileSystem(file, hfileConfig.getHadoopConf());
this.file = HoodieWrapperFileSystem.convertToHoodiePath(file, conf);
this.fs = (HoodieWrapperFileSystem) this.file.getFileSystem(conf);
this.hfileConfig = hfileConfig;
diff --git a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroOrcWriter.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroOrcWriter.java
similarity index 94%
rename from hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroOrcWriter.java
rename to hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroOrcWriter.java
index 3346816125b..07e7bc7f122 100644
--- a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroOrcWriter.java
+++ b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroOrcWriter.java
@@ -18,13 +18,14 @@
package org.apache.hudi.io.storage;
+import org.apache.hudi.avro.HoodieAvroWriteSupport;
import org.apache.hudi.avro.HoodieBloomFilterWriteSupport;
import org.apache.hudi.common.bloom.BloomFilter;
import org.apache.hudi.common.bloom.HoodieDynamicBoundedBloomFilter;
import org.apache.hudi.common.engine.TaskContextSupplier;
-import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.model.HoodieKey;
import org.apache.hudi.common.util.AvroOrcUtils;
+import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.hadoop.fs.HoodieWrapperFileSystem;
import org.apache.hudi.storage.StoragePath;
@@ -45,7 +46,6 @@ import java.nio.ByteBuffer;
import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
-import static org.apache.hudi.avro.HoodieAvroWriteSupport.HOODIE_AVRO_BLOOM_FILTER_METADATA_KEY;
import static org.apache.hudi.common.util.StringUtils.getUTF8Bytes;
public class HoodieAvroOrcWriter implements HoodieAvroFileWriter, Closeable {
@@ -70,7 +70,7 @@ public class HoodieAvroOrcWriter implements HoodieAvroFileWriter, Closeable {
public HoodieAvroOrcWriter(String instantTime, StoragePath file, HoodieOrcConfig config, Schema schema,
TaskContextSupplier taskContextSupplier) throws IOException {
- Configuration conf = FSUtils.registerFileSystem(file, config.getHadoopConf());
+ Configuration conf = HadoopFSUtils.registerFileSystem(file, config.getHadoopConf());
this.file = HoodieWrapperFileSystem.convertToHoodiePath(file, conf);
this.fs = (HoodieWrapperFileSystem) this.file.getFileSystem(conf);
this.instantTime = instantTime;
@@ -152,7 +152,7 @@ public class HoodieAvroOrcWriter implements HoodieAvroFileWriter, Closeable {
if (orcConfig.useBloomFilter()) {
final BloomFilter bloomFilter = orcConfig.getBloomFilter();
- writer.addUserMetadata(HOODIE_AVRO_BLOOM_FILTER_METADATA_KEY, ByteBuffer.wrap(getUTF8Bytes(bloomFilter.serializeToString())));
+ writer.addUserMetadata(HoodieAvroWriteSupport.HOODIE_AVRO_BLOOM_FILTER_METADATA_KEY, ByteBuffer.wrap(getUTF8Bytes(bloomFilter.serializeToString())));
if (minRecordKey != null && maxRecordKey != null) {
writer.addUserMetadata(HoodieBloomFilterWriteSupport.HOODIE_MIN_RECORD_KEY_FOOTER, ByteBuffer.wrap(getUTF8Bytes(minRecordKey)));
writer.addUserMetadata(HoodieBloomFilterWriteSupport.HOODIE_MAX_RECORD_KEY_FOOTER, ByteBuffer.wrap(getUTF8Bytes(maxRecordKey)));
diff --git a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroParquetWriter.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroParquetWriter.java
similarity index 100%
rename from hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroParquetWriter.java
rename to hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/HoodieAvroParquetWriter.java
diff --git a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieBaseParquetWriter.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/HoodieBaseParquetWriter.java
similarity index 97%
rename from hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieBaseParquetWriter.java
rename to hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/HoodieBaseParquetWriter.java
index f237db139ab..06f1e513055 100644
--- a/hudi-common/src/main/java/org/apache/hudi/io/storage/HoodieBaseParquetWriter.java
+++ b/hudi-hadoop-common/src/main/java/org/apache/hudi/io/storage/HoodieBaseParquetWriter.java
@@ -18,8 +18,8 @@
package org.apache.hudi.io.storage;
-import org.apache.hudi.common.fs.FSUtils;
import org.apache.hudi.common.util.VisibleForTesting;
+import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.hadoop.fs.HoodieWrapperFileSystem;
import org.apache.hudi.storage.StoragePath;
@@ -73,7 +73,7 @@ public abstract class HoodieBaseParquetWriter<R> implements Closeable {
parquetWriterbuilder.withDictionaryEncoding(parquetConfig.dictionaryEnabled());
parquetWriterbuilder.withValidation(ParquetWriter.DEFAULT_IS_VALIDATING_ENABLED);
parquetWriterbuilder.withWriterVersion(ParquetWriter.DEFAULT_WRITER_VERSION);
- parquetWriterbuilder.withConf(FSUtils.registerFileSystem(file, parquetConfig.getHadoopConf()));
+ parquetWriterbuilder.withConf(HadoopFSUtils.registerFileSystem(file, parquetConfig.getHadoopConf()));
handleParquetBloomFilters(parquetWriterbuilder, parquetConfig.getHadoopConf());
parquetWriter = parquetWriterbuilder.build();
diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/storage/hadoop/HadoopStorageConfiguration.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/storage/hadoop/HadoopStorageConfiguration.java
index f272f8333eb..ed7b2405247 100644
--- a/hudi-hadoop-common/src/main/java/org/apache/hudi/storage/hadoop/HadoopStorageConfiguration.java
+++ b/hudi-hadoop-common/src/main/java/org/apache/hudi/storage/hadoop/HadoopStorageConfiguration.java
@@ -20,6 +20,7 @@
package org.apache.hudi.storage.hadoop;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.hadoop.fs.inline.HadoopInLineFSUtils;
import org.apache.hudi.storage.StorageConfiguration;
import org.apache.hadoop.conf.Configuration;
@@ -37,8 +38,8 @@ public class HadoopStorageConfiguration extends StorageConfiguration<Configurati
private transient Configuration configuration;
- public HadoopStorageConfiguration() {
- this(new Configuration());
+ public HadoopStorageConfiguration(Boolean loadDefaults) {
+ this(new Configuration(loadDefaults));
}
public HadoopStorageConfiguration(Configuration configuration) {
@@ -82,6 +83,11 @@ public class HadoopStorageConfiguration extends StorageConfiguration<Configurati
return Option.ofNullable(configuration.get(key));
}
+ @Override
+ public StorageConfiguration<Configuration> getInline() {
+ return HadoopInLineFSUtils.buildInlineConf(this);
+ }
+
@Override
public String toString() {
StringBuilder stringBuilder = new StringBuilder();
diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/storage/hadoop/HoodieHadoopStorage.java b/hudi-hadoop-common/src/main/java/org/apache/hudi/storage/hadoop/HoodieHadoopStorage.java
index 1e1ba67ae66..126b17617eb 100644
--- a/hudi-hadoop-common/src/main/java/org/apache/hudi/storage/hadoop/HoodieHadoopStorage.java
+++ b/hudi-hadoop-common/src/main/java/org/apache/hudi/storage/hadoop/HoodieHadoopStorage.java
@@ -19,8 +19,11 @@
package org.apache.hudi.storage.hadoop;
+import org.apache.hudi.common.fs.ConsistencyGuard;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.hadoop.fs.HadoopSeekableDataInputStream;
+import org.apache.hudi.hadoop.fs.HoodieRetryWrapperFileSystem;
+import org.apache.hudi.hadoop.fs.HoodieWrapperFileSystem;
import org.apache.hudi.io.SeekableDataInputStream;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StorageConfiguration;
@@ -43,9 +46,11 @@ import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;
+import static org.apache.hudi.common.util.ValidationUtils.checkArgument;
import static org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToHadoopPath;
import static org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToStoragePath;
import static org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToStoragePathInfo;
+import static org.apache.hudi.hadoop.fs.HadoopFSUtils.getFs;
/**
* Implementation of {@link HoodieStorage} using Hadoop's {@link FileSystem}
@@ -53,6 +58,46 @@ import static org.apache.hudi.hadoop.fs.HadoopFSUtils.convertToStoragePathInfo;
public class HoodieHadoopStorage extends HoodieStorage {
private final FileSystem fs;
+ public HoodieHadoopStorage(HoodieStorage storage) {
+ FileSystem fs = (FileSystem) storage.getFileSystem();
+ if (fs instanceof HoodieWrapperFileSystem) {
+ this.fs = ((HoodieWrapperFileSystem) fs).getFileSystem();
+ } else {
+ this.fs = fs;
+ }
+ }
+
+ public HoodieHadoopStorage(String basePath, Configuration conf) {
+ this(HadoopFSUtils.getFs(basePath, conf));
+ }
+
+ public HoodieHadoopStorage(StoragePath path, StorageConfiguration<?> conf) {
+ this(HadoopFSUtils.getFs(path, conf.unwrapAs(Configuration.class)));
+ }
+
+ public HoodieHadoopStorage(String basePath, StorageConfiguration<?> conf) {
+ this(HadoopFSUtils.getFs(basePath, conf));
+ }
+
+ public HoodieHadoopStorage(StoragePath path,
+ StorageConfiguration<?> conf,
+ boolean enableRetry,
+ long maxRetryIntervalMs,
+ int maxRetryNumbers,
+ long initialRetryIntervalMs,
+ String retryExceptions,
+ ConsistencyGuard consistencyGuard) {
+ FileSystem fileSystem = getFs(path, conf.unwrapCopyAs(Configuration.class));
+
+ if (enableRetry) {
+ fileSystem = new HoodieRetryWrapperFileSystem(fileSystem,
+ maxRetryIntervalMs, maxRetryNumbers, initialRetryIntervalMs, retryExceptions);
+ }
+ checkArgument(!(fileSystem instanceof HoodieWrapperFileSystem),
+ "File System not expected to be that of HoodieWrapperFileSystem");
+ this.fs = new HoodieWrapperFileSystem(fileSystem, consistencyGuard);
+ }
+
public HoodieHadoopStorage(FileSystem fs) {
this.fs = fs;
}
@@ -98,9 +143,9 @@ public class HoodieHadoopStorage extends HoodieStorage {
}
@Override
- public SeekableDataInputStream openSeekable(StoragePath path, int bufferSize) throws IOException {
+ public SeekableDataInputStream openSeekable(StoragePath path, int bufferSize, boolean wrapStream) throws IOException {
return new HadoopSeekableDataInputStream(
- HadoopFSUtils.getFSDataInputStream(fs, path, bufferSize));
+ HadoopFSUtils.getFSDataInputStream(fs, path, bufferSize, wrapStream));
}
@Override
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/bloom/TestBloomFilter.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/bloom/TestBloomFilter.java
similarity index 98%
rename from hudi-common/src/test/java/org/apache/hudi/common/bloom/TestBloomFilter.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/bloom/TestBloomFilter.java
index 2e72b3737a0..cb7d7847694 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/bloom/TestBloomFilter.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/bloom/TestBloomFilter.java
@@ -31,7 +31,7 @@ import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;
-import static org.apache.hudi.common.testutils.FileSystemTestUtils.readLastLineFromResourceFile;
+import static org.apache.hudi.common.testutils.HoodieTestTable.readLastLineFromResourceFile;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/bootstrap/TestBootstrapIndex.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/bootstrap/TestBootstrapIndex.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/bootstrap/TestBootstrapIndex.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/bootstrap/TestBootstrapIndex.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/fs/TestFSUtils.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestFSUtils.java
similarity index 98%
rename from hudi-common/src/test/java/org/apache/hudi/common/fs/TestFSUtils.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestFSUtils.java
index 246fde7aa01..3822535e7db 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/fs/TestFSUtils.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestFSUtils.java
@@ -31,7 +31,6 @@ import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.hadoop.fs.HoodieWrapperFileSystem;
-import org.apache.hudi.hadoop.fs.NoOpConsistencyGuard;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StoragePath;
@@ -96,15 +95,15 @@ public class TestFSUtils extends HoodieCommonTestHarness {
public void testMakeDataFileName() {
String instantTime = HoodieActiveTimeline.formatDate(new Date());
String fileName = UUID.randomUUID().toString();
- assertEquals(FSUtils.makeBaseFileName(instantTime, TEST_WRITE_TOKEN, fileName, BASE_FILE_EXTENSION),
- fileName + "_" + TEST_WRITE_TOKEN + "_" + instantTime + BASE_FILE_EXTENSION);
+ assertEquals(FSUtils.makeBaseFileName(instantTime, TEST_WRITE_TOKEN, fileName, HoodieCommonTestHarness.BASE_FILE_EXTENSION),
+ fileName + "_" + TEST_WRITE_TOKEN + "_" + instantTime + HoodieCommonTestHarness.BASE_FILE_EXTENSION);
}
@Test
public void testMaskFileName() {
String instantTime = HoodieActiveTimeline.formatDate(new Date());
int taskPartitionId = 2;
- assertEquals(FSUtils.maskWithoutFileId(instantTime, taskPartitionId), "*_" + taskPartitionId + "_" + instantTime + BASE_FILE_EXTENSION);
+ assertEquals(FSUtils.maskWithoutFileId(instantTime, taskPartitionId), "*_" + taskPartitionId + "_" + instantTime + HoodieCommonTestHarness.BASE_FILE_EXTENSION);
}
/**
@@ -132,7 +131,7 @@ public class TestFSUtils extends HoodieCommonTestHarness {
"2016/05/16/2_1-0-1_20190528120000",
".hoodie/.temp/2/2016/05/16/2_1-0-1_20190528120000",
".hoodie/.temp/2/2016/04/15/1_1-0-1_20190528120000")
- .map(fileName -> fileName + BASE_FILE_EXTENSION)
+ .map(fileName -> fileName + HoodieCommonTestHarness.BASE_FILE_EXTENSION)
.collect(Collectors.toList());
files.forEach(f -> {
@@ -172,7 +171,7 @@ public class TestFSUtils extends HoodieCommonTestHarness {
public void testGetCommitTime() {
String instantTime = HoodieActiveTimeline.formatDate(new Date());
String fileName = UUID.randomUUID().toString();
- String fullFileName = FSUtils.makeBaseFileName(instantTime, TEST_WRITE_TOKEN, fileName, BASE_FILE_EXTENSION);
+ String fullFileName = FSUtils.makeBaseFileName(instantTime, TEST_WRITE_TOKEN, fileName, HoodieCommonTestHarness.BASE_FILE_EXTENSION);
assertEquals(instantTime, FSUtils.getCommitTime(fullFileName));
// test log file name
fullFileName = FSUtils.makeLogFileName(fileName, HOODIE_LOG.getFileExtension(), instantTime, 1, TEST_WRITE_TOKEN);
@@ -183,7 +182,7 @@ public class TestFSUtils extends HoodieCommonTestHarness {
public void testGetFileNameWithoutMeta() {
String instantTime = HoodieActiveTimeline.formatDate(new Date());
String fileName = UUID.randomUUID().toString();
- String fullFileName = FSUtils.makeBaseFileName(instantTime, TEST_WRITE_TOKEN, fileName, BASE_FILE_EXTENSION);
+ String fullFileName = FSUtils.makeBaseFileName(instantTime, TEST_WRITE_TOKEN, fileName, HoodieCommonTestHarness.BASE_FILE_EXTENSION);
assertEquals(fileName, FSUtils.getFileId(fullFileName));
}
@@ -371,7 +370,7 @@ public class TestFSUtils extends HoodieCommonTestHarness {
final String LOG_EXTENSION = "." + LOG_STR;
// data file name
- String dataFileName = FSUtils.makeBaseFileName(instantTime, writeToken, fileId, BASE_FILE_EXTENSION);
+ String dataFileName = FSUtils.makeBaseFileName(instantTime, writeToken, fileId, HoodieCommonTestHarness.BASE_FILE_EXTENSION);
assertEquals(instantTime, FSUtils.getCommitTime(dataFileName));
assertEquals(fileId, FSUtils.getFileId(dataFileName));
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsWithRetryWrapperEnable.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsWithRetryWrapperEnable.java
similarity index 99%
rename from hudi-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsWithRetryWrapperEnable.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsWithRetryWrapperEnable.java
index 2ee65d6f045..2093e658c4e 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsWithRetryWrapperEnable.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestFSUtilsWithRetryWrapperEnable.java
@@ -21,7 +21,6 @@ package org.apache.hudi.common.fs;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.hadoop.fs.HoodieRetryWrapperFileSystem;
import org.apache.hudi.hadoop.fs.HoodieWrapperFileSystem;
-import org.apache.hudi.hadoop.fs.NoOpConsistencyGuard;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StoragePath;
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/fs/TestHoodieWrapperFileSystem.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestHoodieWrapperFileSystem.java
similarity index 98%
rename from hudi-common/src/test/java/org/apache/hudi/common/fs/TestHoodieWrapperFileSystem.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestHoodieWrapperFileSystem.java
index 93a321166c0..c7b5217524e 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/fs/TestHoodieWrapperFileSystem.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestHoodieWrapperFileSystem.java
@@ -23,7 +23,6 @@ import org.apache.hudi.common.testutils.minicluster.HdfsTestService;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.hadoop.fs.HoodieWrapperFileSystem;
-import org.apache.hudi.hadoop.fs.NoOpConsistencyGuard;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StoragePath;
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/fs/TestStorageSchemes.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestStorageSchemes.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/fs/TestStorageSchemes.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/TestStorageSchemes.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/fs/inline/InLineFSUtilsTest.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/inline/InLineFSUtilsTest.java
similarity index 93%
rename from hudi-common/src/test/java/org/apache/hudi/common/fs/inline/InLineFSUtilsTest.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/inline/InLineFSUtilsTest.java
index 04eefcf15dd..f46a8d23f25 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/fs/inline/InLineFSUtilsTest.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/inline/InLineFSUtilsTest.java
@@ -19,9 +19,8 @@
package org.apache.hudi.common.fs.inline;
import org.apache.hudi.common.testutils.FileSystemTestUtils;
-import org.apache.hudi.hadoop.fs.inline.InLineFSUtils;
-import org.apache.hudi.hadoop.fs.inline.InLineFileSystem;
import org.apache.hudi.storage.StoragePath;
+import org.apache.hudi.storage.inline.InLineFSUtils;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
@@ -32,7 +31,7 @@ import java.util.stream.Stream;
import static org.junit.jupiter.api.Assertions.assertEquals;
/**
- * Tests {@link InLineFileSystem}.
+ * Tests {@link InLineFSUtils}.
*/
public class InLineFSUtilsTest {
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/fs/inline/TestInLineFileSystem.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/inline/TestInLineFileSystem.java
similarity index 98%
rename from hudi-common/src/test/java/org/apache/hudi/common/fs/inline/TestInLineFileSystem.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/inline/TestInLineFileSystem.java
index dd9bdc8cc49..76b55429024 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/fs/inline/TestInLineFileSystem.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/inline/TestInLineFileSystem.java
@@ -20,7 +20,7 @@ package org.apache.hudi.common.fs.inline;
import org.apache.hudi.common.testutils.FileSystemTestUtils;
import org.apache.hudi.common.util.collection.Pair;
-import org.apache.hudi.hadoop.fs.inline.InLineFSUtils;
+import org.apache.hudi.hadoop.fs.inline.HadoopInLineFSUtils;
import org.apache.hudi.hadoop.fs.inline.InLineFileSystem;
import org.apache.hudi.storage.StoragePath;
@@ -350,12 +350,12 @@ public class TestInLineFileSystem {
if (inputPath.toString().contains(":")) {
scheme = inputPath.toString().split(":")[0];
}
- final StoragePath actualInLineFSPath = InLineFSUtils.getInlineFilePath(
+ final StoragePath actualInLineFSPath = HadoopInLineFSUtils.getInlineFilePath(
new StoragePath(inputPath.toUri()), scheme, 10, 10);
assertEquals(expectedInLineFSPath, actualInLineFSPath);
final StoragePath actualOuterFilePath =
- InLineFSUtils.getOuterFilePathFromInlinePath(actualInLineFSPath);
+ HadoopInLineFSUtils.getOuterFilePathFromInlinePath(actualInLineFSPath);
assertEquals(expectedTransformedInputPath, actualOuterFilePath);
}
}
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/fs/inline/TestInLineFileSystemHFileInLiningBase.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/inline/TestInLineFileSystemHFileInLiningBase.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/fs/inline/TestInLineFileSystemHFileInLiningBase.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/inline/TestInLineFileSystemHFileInLiningBase.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/fs/inline/TestInLineFileSystemWithHBaseHFileReader.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/inline/TestInLineFileSystemWithHBaseHFileReader.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/fs/inline/TestInLineFileSystemWithHBaseHFileReader.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/inline/TestInLineFileSystemWithHBaseHFileReader.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/fs/inline/TestInLineFileSystemWithHFileReader.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/inline/TestInLineFileSystemWithHFileReader.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/fs/inline/TestInLineFileSystemWithHFileReader.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/inline/TestInLineFileSystemWithHFileReader.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/fs/inline/TestInMemoryFileSystem.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/inline/TestInMemoryFileSystem.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/fs/inline/TestInMemoryFileSystem.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/inline/TestInMemoryFileSystem.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/fs/inline/TestParquetInLining.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/inline/TestParquetInLining.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/fs/inline/TestParquetInLining.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/fs/inline/TestParquetInLining.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormat.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormat.java
similarity index 99%
rename from hudi-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormat.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormat.java
index c49e804c31a..3713950eb2b 100755
--- a/hudi-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormat.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormat.java
@@ -61,7 +61,6 @@ import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.collection.ClosableIterator;
import org.apache.hudi.common.util.collection.ExternalSpillableMap;
import org.apache.hudi.exception.CorruptedLogFileException;
-import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StoragePath;
@@ -71,7 +70,6 @@ import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.generic.IndexedRecord;
-import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocalFileSystem;
@@ -436,7 +434,7 @@ public class TestHoodieLogFormat extends HoodieCommonTestHarness {
header.put(HoodieLogBlock.HeaderMetadataType.SCHEMA, getSimpleSchema().toString());
byte[] dataBlockContentBytes = getDataBlock(DEFAULT_DATA_BLOCK_TYPE, records, header).getContentBytes();
HoodieLogBlock.HoodieLogBlockContentLocation logBlockContentLoc = new HoodieLogBlock.HoodieLogBlockContentLocation(
- HadoopFSUtils.getStorageConf(new Configuration()), null, 0, dataBlockContentBytes.length, 0);
+ HoodieTestUtils.getDefaultStorageConfWithDefaults(), null, 0, dataBlockContentBytes.length, 0);
HoodieDataBlock reusableDataBlock = new HoodieAvroDataBlock(null, Option.ofNullable(dataBlockContentBytes), false,
logBlockContentLoc, Option.ofNullable(getSimpleSchema()), header, new HashMap<>(), HoodieRecord.RECORD_KEY_METADATA_FIELD);
long writtenSize = 0;
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormatAppendFailure.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormatAppendFailure.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormatAppendFailure.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/functional/TestHoodieLogFormatAppendFailure.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/model/TestHoodieFileGroup.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/model/TestHoodieFileGroup.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/model/TestHoodieFileGroup.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/model/TestHoodieFileGroup.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/model/TestHoodiePartitionMetadata.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/model/TestHoodiePartitionMetadata.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/model/TestHoodiePartitionMetadata.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/model/TestHoodiePartitionMetadata.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/model/TestHoodieRecord.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/model/TestHoodieRecord.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/model/TestHoodieRecord.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/model/TestHoodieRecord.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/model/TestHoodieRecordDelegate.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/model/TestHoodieRecordDelegate.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/model/TestHoodieRecordDelegate.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/model/TestHoodieRecordDelegate.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfig.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfig.java
similarity index 97%
rename from hudi-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfig.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfig.java
index c9ac1c0c9a6..297ddda2091 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfig.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableConfig.java
@@ -19,14 +19,13 @@
package org.apache.hudi.common.table;
import org.apache.hudi.common.testutils.HoodieCommonTestHarness;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
import org.apache.hudi.common.util.CollectionUtils;
import org.apache.hudi.exception.HoodieIOException;
-import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StoragePath;
-import org.apache.hadoop.conf.Configuration;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -62,7 +61,7 @@ public class TestHoodieTableConfig extends HoodieCommonTestHarness {
@BeforeEach
public void setUp() throws Exception {
initPath();
- storage = HoodieStorageUtils.getStorage(basePath, HadoopFSUtils.getStorageConf(new Configuration()));
+ storage = HoodieStorageUtils.getStorage(basePath, HoodieTestUtils.getDefaultStorageConfWithDefaults());
metaPath = new StoragePath(basePath, HoodieTableMetaClient.METAFOLDER_NAME);
Properties props = new Properties();
props.setProperty(HoodieTableConfig.NAME.key(), "test-table");
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableMetaClient.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableMetaClient.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableMetaClient.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestHoodieTableMetaClient.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/TestTableSchemaResolver.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestTableSchemaResolver.java
similarity index 95%
rename from hudi-common/src/test/java/org/apache/hudi/common/table/TestTableSchemaResolver.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestTableSchemaResolver.java
index a4801fa5464..76ac5e7abe9 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/table/TestTableSchemaResolver.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestTableSchemaResolver.java
@@ -24,9 +24,9 @@ import org.apache.hudi.common.table.log.HoodieLogFormat;
import org.apache.hudi.common.table.log.block.HoodieDataBlock;
import org.apache.hudi.common.table.log.block.HoodieLogBlock;
import org.apache.hudi.common.testutils.HoodieTestDataGenerator;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
import org.apache.hudi.common.testutils.SchemaTestUtil;
import org.apache.hudi.common.util.Option;
-import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.internal.schema.HoodieSchemaException;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.HoodieStorageUtils;
@@ -34,7 +34,6 @@ import org.apache.hudi.storage.StoragePath;
import org.apache.avro.Schema;
import org.apache.avro.generic.IndexedRecord;
-import org.apache.hadoop.conf.Configuration;
import org.apache.parquet.avro.AvroSchemaConverter;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
@@ -100,7 +99,7 @@ public class TestTableSchemaResolver {
assertEquals(
new AvroSchemaConverter().convert(expectedSchema),
TableSchemaResolver.readSchemaFromLogFile(HoodieStorageUtils.getStorage(
- logFilePath, HadoopFSUtils.getStorageConf(new Configuration())), logFilePath));
+ logFilePath, HoodieTestUtils.getDefaultStorageConfWithDefaults()), logFilePath));
}
private String initTestDir(String folderName) throws IOException {
@@ -111,7 +110,7 @@ public class TestTableSchemaResolver {
private StoragePath writeLogFile(StoragePath partitionPath, Schema schema) throws IOException, URISyntaxException, InterruptedException {
HoodieStorage storage = HoodieStorageUtils.getStorage(
- partitionPath, HadoopFSUtils.getStorageConf(new Configuration()));
+ partitionPath, HoodieTestUtils.getDefaultStorageConfWithDefaults());
HoodieLogFormat.Writer writer =
HoodieLogFormat.newWriterBuilder().onParentPath(partitionPath).withFileExtension(HoodieLogFile.DELTA_EXTENSION)
.withFileId("test-fileid1").overBaseCommit("100").withStorage(storage).build();
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/TestTimelineUtils.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/log/TestLogReaderUtils.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/log/TestLogReaderUtils.java
similarity index 97%
rename from hudi-common/src/test/java/org/apache/hudi/common/table/log/TestLogReaderUtils.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/log/TestLogReaderUtils.java
index fd8e3a5cd28..154f2b22941 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/table/log/TestLogReaderUtils.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/log/TestLogReaderUtils.java
@@ -32,7 +32,7 @@ import java.util.Random;
import java.util.Set;
import java.util.stream.Collectors;
-import static org.apache.hudi.common.testutils.FileSystemTestUtils.readLastLineFromResourceFile;
+import static org.apache.hudi.common.testutils.HoodieTestTable.readLastLineFromResourceFile;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieActiveTimeline.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieActiveTimeline.java
similarity index 99%
rename from hudi-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieActiveTimeline.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieActiveTimeline.java
index 4435707e78f..a317d616136 100755
--- a/hudi-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieActiveTimeline.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieActiveTimeline.java
@@ -18,6 +18,7 @@
package org.apache.hudi.common.table.timeline;
+import org.apache.hudi.common.fs.NoOpConsistencyGuard;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieInstant.State;
import org.apache.hudi.common.table.timeline.versioning.TimelineLayoutVersion;
@@ -27,7 +28,6 @@ import org.apache.hudi.common.util.CollectionUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.hadoop.fs.HoodieWrapperFileSystem;
-import org.apache.hudi.hadoop.fs.NoOpConsistencyGuard;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StoragePath;
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieInstant.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieInstant.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieInstant.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/timeline/TestHoodieInstant.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFSViewWithClustering.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFSViewWithClustering.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFSViewWithClustering.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFSViewWithClustering.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFileSystemView.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFileSystemView.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFileSystemView.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestHoodieTableFileSystemView.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/view/TestIncrementalFSViewSync.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestIncrementalFSViewSync.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/table/view/TestIncrementalFSViewSync.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestIncrementalFSViewSync.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/view/TestRocksDBBasedIncrementalFSViewSync.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestRocksDBBasedIncrementalFSViewSync.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/table/view/TestRocksDBBasedIncrementalFSViewSync.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestRocksDBBasedIncrementalFSViewSync.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/view/TestRocksDbBasedFileSystemView.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestRocksDbBasedFileSystemView.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/table/view/TestRocksDbBasedFileSystemView.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestRocksDbBasedFileSystemView.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/view/TestSpillableMapBasedFileSystemView.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestSpillableMapBasedFileSystemView.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/table/view/TestSpillableMapBasedFileSystemView.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestSpillableMapBasedFileSystemView.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/table/view/TestSpillableMapBasedIncrementalFSViewSync.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestSpillableMapBasedIncrementalFSViewSync.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/table/view/TestSpillableMapBasedIncrementalFSViewSync.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/table/view/TestSpillableMapBasedIncrementalFSViewSync.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/testutils/FileSystemTestUtils.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/FileSystemTestUtils.java
similarity index 61%
rename from hudi-common/src/test/java/org/apache/hudi/common/testutils/FileSystemTestUtils.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/FileSystemTestUtils.java
index 232c14cc31c..162740b55a1 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/testutils/FileSystemTestUtils.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/FileSystemTestUtils.java
@@ -18,26 +18,15 @@
package org.apache.hudi.common.testutils;
-import org.apache.hudi.common.table.log.TestLogReaderUtils;
-import org.apache.hudi.common.util.FileIOUtils;
-import org.apache.hudi.hadoop.fs.inline.InLineFSUtils;
import org.apache.hudi.hadoop.fs.inline.InLineFileSystem;
import org.apache.hudi.hadoop.fs.inline.InMemoryFileSystem;
-import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.StoragePath;
-import org.apache.hudi.storage.StoragePathInfo;
+import org.apache.hudi.storage.inline.InLineFSUtils;
-import org.apache.hadoop.fs.FileStatus;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.RemoteIterator;
import java.io.File;
import java.io.IOException;
-import java.io.InputStream;
-import java.util.ArrayList;
-import java.util.List;
import java.util.Random;
import java.util.UUID;
@@ -79,34 +68,4 @@ public class FileSystemTestUtils {
throw new IOException(message);
}
}
-
- public static List<FileStatus> listRecursive(FileSystem fs, Path path) throws IOException {
- return listFiles(fs, path, true);
- }
-
- public static List<FileStatus> listFiles(FileSystem fs, Path path, boolean recursive) throws IOException {
- RemoteIterator<LocatedFileStatus> itr = fs.listFiles(path, recursive);
- List<FileStatus> statuses = new ArrayList<>();
- while (itr.hasNext()) {
- statuses.add(itr.next());
- }
- return statuses;
- }
-
- public static List<StoragePathInfo> listRecursive(HoodieStorage storage, StoragePath path)
- throws IOException {
- return listFiles(storage, path);
- }
-
- public static List<StoragePathInfo> listFiles(HoodieStorage storage, StoragePath path)
- throws IOException {
- return storage.listFiles(path);
- }
-
- public static String readLastLineFromResourceFile(String resourceName) throws IOException {
- try (InputStream inputStream = TestLogReaderUtils.class.getResourceAsStream(resourceName)) {
- List<String> lines = FileIOUtils.readAsUTFStringLines(inputStream);
- return lines.get(lines.size() - 1);
- }
- }
}
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieCommonTestHarness.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/HoodieCommonTestHarness.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieCommonTestHarness.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/HoodieCommonTestHarness.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestTable.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestTable.java
similarity index 97%
rename from hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestTable.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestTable.java
index 8781765702c..1192004c9e9 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestTable.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/HoodieTestTable.java
@@ -51,12 +51,14 @@ import org.apache.hudi.common.model.IOType;
import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.table.log.TestLogReaderUtils;
import org.apache.hudi.common.table.timeline.HoodieActiveTimeline;
import org.apache.hudi.common.table.timeline.HoodieInstant;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.table.timeline.TimelineMetadataUtils;
import org.apache.hudi.common.table.timeline.versioning.clean.CleanPlanV2MigrationHandler;
import org.apache.hudi.common.util.CompactionUtils;
+import org.apache.hudi.common.util.FileIOUtils;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.StringUtils;
import org.apache.hudi.common.util.ValidationUtils;
@@ -68,11 +70,14 @@ import org.apache.hudi.storage.StoragePathInfo;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.RemoteIterator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
+import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.time.Instant;
@@ -783,17 +788,47 @@ public class HoodieTestTable {
}
public List<StoragePathInfo> listAllBaseFiles(String fileExtension) throws IOException {
- return FileSystemTestUtils.listRecursive(storage, new StoragePath(basePath)).stream()
+ return listRecursive(storage, new StoragePath(basePath)).stream()
.filter(fileInfo -> fileInfo.getPath().getName().endsWith(fileExtension))
.collect(Collectors.toList());
}
+ public static List<FileStatus> listRecursive(FileSystem fs, Path path) throws IOException {
+ return listFiles(fs, path, true);
+ }
+
+ public static List<FileStatus> listFiles(FileSystem fs, Path path, boolean recursive) throws IOException {
+ RemoteIterator<LocatedFileStatus> itr = fs.listFiles(path, recursive);
+ List<FileStatus> statuses = new ArrayList<>();
+ while (itr.hasNext()) {
+ statuses.add(itr.next());
+ }
+ return statuses;
+ }
+
+ public static List<StoragePathInfo> listRecursive(HoodieStorage storage, StoragePath path)
+ throws IOException {
+ return listFiles(storage, path);
+ }
+
+ public static List<StoragePathInfo> listFiles(HoodieStorage storage, StoragePath path)
+ throws IOException {
+ return storage.listFiles(path);
+ }
+
+ public static String readLastLineFromResourceFile(String resourceName) throws IOException {
+ try (InputStream inputStream = TestLogReaderUtils.class.getResourceAsStream(resourceName)) {
+ List<String> lines = FileIOUtils.readAsUTFStringLines(inputStream);
+ return lines.get(lines.size() - 1);
+ }
+ }
+
public List<StoragePathInfo> listAllLogFiles() throws IOException {
return listAllLogFiles(HoodieFileFormat.HOODIE_LOG.getFileExtension());
}
public List<StoragePathInfo> listAllLogFiles(String fileExtension) throws IOException {
- return FileSystemTestUtils.listRecursive(storage, new StoragePath(basePath)).stream()
+ return listRecursive(storage, new StoragePath(basePath)).stream()
.filter(
fileInfo -> !fileInfo.getPath().toString()
.contains(HoodieTableMetaClient.METAFOLDER_NAME))
@@ -808,7 +843,7 @@ public class HoodieTestTable {
}
public FileStatus[] listAllFilesInPartition(String partitionPath) throws IOException {
- return FileSystemTestUtils.listRecursive(fs,
+ return listRecursive(fs,
new Path(Paths.get(basePath, partitionPath).toString())).stream()
.filter(entry -> {
boolean toReturn = true;
@@ -831,7 +866,7 @@ public class HoodieTestTable {
}
public FileStatus[] listAllFilesInTempFolder() throws IOException {
- return FileSystemTestUtils.listRecursive(fs, new Path(Paths.get(basePath, HoodieTableMetaClient.TEMPFOLDER_NAME).toString())).toArray(new FileStatus[0]);
+ return listRecursive(fs, new Path(Paths.get(basePath, HoodieTableMetaClient.TEMPFOLDER_NAME).toString())).toArray(new FileStatus[0]);
}
public void deleteFilesInPartition(String partitionPath, List<String> filesToDelete) throws IOException {
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/testutils/minicluster/HdfsTestService.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/minicluster/HdfsTestService.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/testutils/minicluster/HdfsTestService.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/testutils/minicluster/HdfsTestService.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/util/TestAvroOrcUtils.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestAvroOrcUtils.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/util/TestAvroOrcUtils.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestAvroOrcUtils.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/util/TestClusteringUtils.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestClusteringUtils.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/util/TestClusteringUtils.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestClusteringUtils.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/util/TestCommitUtils.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestCommitUtils.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/util/TestCommitUtils.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestCommitUtils.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/util/TestCompactionUtils.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestCompactionUtils.java
similarity index 99%
rename from hudi-common/src/test/java/org/apache/hudi/common/util/TestCompactionUtils.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestCompactionUtils.java
index 32dfcecbcbb..4741cdef1f8 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/util/TestCompactionUtils.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestCompactionUtils.java
@@ -32,7 +32,6 @@ import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.table.timeline.versioning.compaction.CompactionPlanMigrator;
import org.apache.hudi.common.testutils.CompactionTestUtils.DummyHoodieBaseFile;
import org.apache.hudi.common.testutils.HoodieCommonTestHarness;
-import org.apache.hudi.common.testutils.HoodieTestUtils;
import org.apache.hudi.common.util.collection.Pair;
import org.apache.hudi.storage.StoragePath;
@@ -217,7 +216,7 @@ public class TestCompactionUtils extends HoodieCommonTestHarness {
// schedule similar plan again so that there will be duplicates
plan1.getOperations().get(0).setDataFilePath("bla");
scheduleCompaction(metaClient, "005", plan1);
- metaClient = HoodieTestUtils.createMetaClient(metaClient.getStorageConf(), basePath);
+ metaClient = createMetaClient(metaClient.getStorageConf(), basePath);
assertThrows(IllegalStateException.class, () -> {
CompactionUtils.getAllPendingCompactionOperations(metaClient);
});
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/util/TestDFSPropertiesConfiguration.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestDFSPropertiesConfiguration.java
similarity index 96%
rename from hudi-common/src/test/java/org/apache/hudi/common/util/TestDFSPropertiesConfiguration.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestDFSPropertiesConfiguration.java
index 21412696f2c..f6caa31a62c 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/util/TestDFSPropertiesConfiguration.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestDFSPropertiesConfiguration.java
@@ -21,6 +21,7 @@ package org.apache.hudi.common.util;
import org.apache.hudi.common.config.DFSPropertiesConfiguration;
import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
import org.apache.hudi.common.testutils.minicluster.HdfsTestService;
import org.apache.hudi.exception.HoodieIOException;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
@@ -42,8 +43,6 @@ import java.io.File;
import java.io.IOException;
import java.io.PrintStream;
-import static org.apache.hudi.common.testutils.HoodieTestUtils.shouldUseExternalHdfs;
-import static org.apache.hudi.common.testutils.HoodieTestUtils.useExternalHdfs;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -65,8 +64,8 @@ public class TestDFSPropertiesConfiguration {
@BeforeAll
public static void initClass() throws Exception {
- if (shouldUseExternalHdfs()) {
- dfs = useExternalHdfs();
+ if (HoodieTestUtils.shouldUseExternalHdfs()) {
+ dfs = HoodieTestUtils.useExternalHdfs();
} else {
hdfsTestService = new HdfsTestService();
dfsCluster = hdfsTestService.start(true);
@@ -169,7 +168,7 @@ public class TestDFSPropertiesConfiguration {
String.format(
"file:%s",
getClass().getClassLoader()
- .getResource("props/test.properties")
+ .getResource("props/testdfs.properties")
.getPath()
)
));
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/util/TestFileIOUtils.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestFileIOUtils.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/util/TestFileIOUtils.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestFileIOUtils.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/util/TestMarkerUtils.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestMarkerUtils.java
similarity index 94%
rename from hudi-common/src/test/java/org/apache/hudi/common/util/TestMarkerUtils.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestMarkerUtils.java
index c604d276ba9..085a981b220 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/util/TestMarkerUtils.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestMarkerUtils.java
@@ -20,13 +20,12 @@ package org.apache.hudi.common.util;
import org.apache.hudi.common.table.marker.MarkerType;
import org.apache.hudi.common.testutils.HoodieCommonTestHarness;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
import org.apache.hudi.exception.HoodieException;
-import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StoragePath;
-import org.apache.hadoop.conf.Configuration;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -43,8 +42,7 @@ class TestMarkerUtils extends HoodieCommonTestHarness {
@BeforeEach
public void setup() {
initPath();
- storage = HoodieStorageUtils.getStorage(
- basePath, HadoopFSUtils.getStorageConf(new Configuration()));
+ storage = HoodieStorageUtils.getStorage(basePath, HoodieTestUtils.getDefaultStorageConfWithDefaults());
}
@Test
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/util/TestParquetUtils.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestParquetUtils.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/util/TestParquetUtils.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestParquetUtils.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/util/TestTablePathUtils.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestTablePathUtils.java
similarity index 96%
rename from hudi-common/src/test/java/org/apache/hudi/common/util/TestTablePathUtils.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestTablePathUtils.java
index 95b08d9d620..94943a436ee 100644
--- a/hudi-common/src/test/java/org/apache/hudi/common/util/TestTablePathUtils.java
+++ b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/TestTablePathUtils.java
@@ -21,12 +21,11 @@ import org.apache.hudi.common.model.HoodieFileFormat;
import org.apache.hudi.common.model.HoodiePartitionMetadata;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.HoodieTableMetaClient;
-import org.apache.hudi.hadoop.fs.HadoopFSUtils;
+import org.apache.hudi.common.testutils.HoodieTestUtils;
import org.apache.hudi.storage.HoodieStorage;
import org.apache.hudi.storage.HoodieStorageUtils;
import org.apache.hudi.storage.StoragePath;
-import org.apache.hadoop.conf.Configuration;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
@@ -64,8 +63,7 @@ public final class TestTablePathUtils {
private void setup(Option<HoodieFileFormat> partitionMetafileFormat) throws IOException {
URI tablePathURI = Paths.get(tempDir.getAbsolutePath(), "test_table").toUri();
tablePath = new StoragePath(tablePathURI);
- storage = HoodieStorageUtils.getStorage(
- tablePathURI.toString(), HadoopFSUtils.getStorageConf(new Configuration()));
+ storage = HoodieStorageUtils.getStorage(tablePathURI.toString(), HoodieTestUtils.getDefaultStorageConfWithDefaults());
// Create bootstrap index folder
assertTrue(new File(
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/util/collection/TestBitCaskDiskMap.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/collection/TestBitCaskDiskMap.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/util/collection/TestBitCaskDiskMap.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/collection/TestBitCaskDiskMap.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/util/collection/TestExternalSpillableMap.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/collection/TestExternalSpillableMap.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/util/collection/TestExternalSpillableMap.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/collection/TestExternalSpillableMap.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/util/collection/TestRocksDbBasedMap.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/collection/TestRocksDbBasedMap.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/util/collection/TestRocksDbBasedMap.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/collection/TestRocksDbBasedMap.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/util/collection/TestRocksDbDiskMap.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/collection/TestRocksDbDiskMap.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/util/collection/TestRocksDbDiskMap.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/common/util/collection/TestRocksDbDiskMap.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/internal/schema/io/TestFileBasedInternalSchemaStorageManager.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/internal/schema/io/TestFileBasedInternalSchemaStorageManager.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/internal/schema/io/TestFileBasedInternalSchemaStorageManager.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/internal/schema/io/TestFileBasedInternalSchemaStorageManager.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/io/storage/TestHoodieAvroFileReaderFactory.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/TestHoodieAvroFileReaderFactory.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/io/storage/TestHoodieAvroFileReaderFactory.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/TestHoodieAvroFileReaderFactory.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/io/storage/TestHoodieBaseParquetWriter.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/TestHoodieBaseParquetWriter.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/io/storage/TestHoodieBaseParquetWriter.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/TestHoodieBaseParquetWriter.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/io/storage/TestHoodieHBaseHFileReaderWriter.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/TestHoodieHBaseHFileReaderWriter.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/io/storage/TestHoodieHBaseHFileReaderWriter.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/TestHoodieHBaseHFileReaderWriter.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/io/storage/TestHoodieHFileReaderWriter.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/TestHoodieHFileReaderWriter.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/io/storage/TestHoodieHFileReaderWriter.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/TestHoodieHFileReaderWriter.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/io/storage/TestHoodieHFileReaderWriterBase.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/TestHoodieHFileReaderWriterBase.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/io/storage/TestHoodieHFileReaderWriterBase.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/TestHoodieHFileReaderWriterBase.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/io/storage/TestHoodieOrcReaderWriter.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/TestHoodieOrcReaderWriter.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/io/storage/TestHoodieOrcReaderWriter.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/TestHoodieOrcReaderWriter.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/io/storage/TestHoodieReaderWriterBase.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/TestHoodieReaderWriterBase.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/io/storage/TestHoodieReaderWriterBase.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/io/storage/TestHoodieReaderWriterBase.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/metadata/TestFileSystemBackedTableMetadata.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/metadata/TestFileSystemBackedTableMetadata.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/metadata/TestFileSystemBackedTableMetadata.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/metadata/TestFileSystemBackedTableMetadata.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/metadata/TestHoodieMetadataPayload.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/metadata/TestHoodieMetadataPayload.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/metadata/TestHoodieMetadataPayload.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/metadata/TestHoodieMetadataPayload.java
diff --git a/hudi-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java b/hudi-hadoop-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java
rename to hudi-hadoop-common/src/test/java/org/apache/hudi/metadata/TestHoodieTableMetadataUtil.java
diff --git a/hudi-common/src/test/resources/external-config/hudi-defaults.conf b/hudi-hadoop-common/src/test/resources/external-config/hudi-defaults.conf
similarity index 100%
rename from hudi-common/src/test/resources/external-config/hudi-defaults.conf
rename to hudi-hadoop-common/src/test/resources/external-config/hudi-defaults.conf
diff --git a/hudi-hadoop-common/src/test/resources/props/testdfs.properties b/hudi-hadoop-common/src/test/resources/props/testdfs.properties
new file mode 100644
index 00000000000..0e9f3e7aa27
--- /dev/null
+++ b/hudi-hadoop-common/src/test/resources/props/testdfs.properties
@@ -0,0 +1,17 @@
+# 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.
+
+some.random.prop=123
\ No newline at end of file
diff --git a/hudi-hadoop-mr/pom.xml b/hudi-hadoop-mr/pom.xml
index 2b0ffd90fef..dec8ea5812a 100644
--- a/hudi-hadoop-mr/pom.xml
+++ b/hudi-hadoop-mr/pom.xml
@@ -44,6 +44,12 @@
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
<!-- Kryo -->
<dependency>
<groupId>com.esotericsoftware</groupId>
@@ -108,6 +114,22 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-io</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
</dependencies>
diff --git a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/realtime/TestHoodieMergeOnReadSnapshotReader.java b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/realtime/TestHoodieMergeOnReadSnapshotReader.java
index b326e7f62d9..86f7f6c82a8 100644
--- a/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/realtime/TestHoodieMergeOnReadSnapshotReader.java
+++ b/hudi-hadoop-mr/src/test/java/org/apache/hudi/hadoop/realtime/TestHoodieMergeOnReadSnapshotReader.java
@@ -62,7 +62,6 @@ import java.util.Map;
import java.util.stream.Collectors;
import static org.apache.hudi.common.fs.FSUtils.getRelativePartitionPath;
-import static org.apache.hudi.hadoop.fs.HadoopFSUtils.getFs;
import static org.apache.hudi.hadoop.testutils.InputFormatTestUtil.writeDataBlockToLogFile;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -90,7 +89,7 @@ public class TestHoodieMergeOnReadSnapshotReader {
baseJobConf.set(HoodieRealtimeConfig.MAX_DFS_STREAM_BUFFER_SIZE_PROP, String.valueOf(1024 * 1024));
baseJobConf.set(serdeConstants.LIST_COLUMNS, COLUMNS);
baseJobConf.set(serdeConstants.LIST_COLUMN_TYPES, COLUMN_TYPES);
- storage = HoodieStorageUtils.getStorage(getFs(basePath.toUri().toString(), baseJobConf));
+ storage = HoodieStorageUtils.getStorage(basePath.toUri().toString(), baseJobConf);
}
@AfterEach
@@ -114,7 +113,7 @@ public class TestHoodieMergeOnReadSnapshotReader {
private void testReaderInternal(boolean partitioned, HoodieLogBlock.HoodieLogBlockType logBlockType) throws Exception {
// initial commit
Schema schema = HoodieAvroUtils.addMetadataFields(SchemaTestUtil.getEvolvedSchema());
- HoodieTestUtils.init(HadoopFSUtils.getStorageConf(hadoopConf), basePath.toString(), HoodieTableType.MERGE_ON_READ);
+ HoodieTestUtils.init(HoodieStorageUtils.getStorageConf(hadoopConf), basePath.toString(), HoodieTableType.MERGE_ON_READ);
String baseInstant = "100";
File partitionDir = partitioned ? InputFormatTestUtil.prepareParquetTable(basePath, schema, 1, TOTAL_RECORDS, baseInstant,
HoodieTableType.MERGE_ON_READ)
diff --git a/hudi-integ-test/pom.xml b/hudi-integ-test/pom.xml
index 64ed135fba0..7ab1271dca9 100644
--- a/hudi-integ-test/pom.xml
+++ b/hudi-integ-test/pom.xml
@@ -210,6 +210,20 @@
<classifier>tests</classifier>
<type>test-jar</type>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-io</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-spark_${scala.binary.version}</artifactId>
diff --git a/hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/writer/AvroFileDeltaInputWriter.java b/hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/writer/AvroFileDeltaInputWriter.java
index efc40437b8e..e9149e8aaa5 100644
--- a/hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/writer/AvroFileDeltaInputWriter.java
+++ b/hudi-integ-test/src/main/java/org/apache/hudi/integ/testsuite/writer/AvroFileDeltaInputWriter.java
@@ -18,7 +18,7 @@
package org.apache.hudi.integ.testsuite.writer;
-import org.apache.hudi.common.fs.FSUtils;
+import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.hadoop.fs.HoodieWrapperFileSystem;
import org.apache.hudi.storage.StoragePath;
@@ -71,7 +71,7 @@ public class AvroFileDeltaInputWriter implements DeltaInputWriter<GenericRecord>
StoragePath path = new StoragePath(basePath, UUID.randomUUID().toString() + AVRO_EXTENSION);
this.file = HoodieWrapperFileSystem.convertToHoodiePath(path, configuration);
this.fs = (HoodieWrapperFileSystem) this.file
- .getFileSystem(FSUtils.registerFileSystem(path, configuration));
+ .getFileSystem(HadoopFSUtils.registerFileSystem(path, configuration));
this.output = this.fs.create(this.file);
this.writer = new GenericDatumWriter(schema);
this.dataFileWriter = new DataFileWriter<>(writer).create(schema, output);
diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/ConsistencyGuard.java b/hudi-io/src/main/java/org/apache/hudi/common/fs/ConsistencyGuard.java
similarity index 98%
rename from hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/ConsistencyGuard.java
rename to hudi-io/src/main/java/org/apache/hudi/common/fs/ConsistencyGuard.java
index ac615fb1048..e475a9195cc 100644
--- a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/ConsistencyGuard.java
+++ b/hudi-io/src/main/java/org/apache/hudi/common/fs/ConsistencyGuard.java
@@ -17,7 +17,7 @@
* under the License.
*/
-package org.apache.hudi.hadoop.fs;
+package org.apache.hudi.common.fs;
import org.apache.hudi.storage.StoragePath;
diff --git a/hudi-io/src/main/java/org/apache/hudi/storage/HoodieStorage.java b/hudi-io/src/main/java/org/apache/hudi/storage/HoodieStorage.java
index b7e98776043..586b5b0a56f 100644
--- a/hudi-io/src/main/java/org/apache/hudi/storage/HoodieStorage.java
+++ b/hudi-io/src/main/java/org/apache/hudi/storage/HoodieStorage.java
@@ -123,11 +123,12 @@ public abstract class HoodieStorage implements Closeable {
*
* @param path the file to open.
* @param bufferSize buffer size to use.
+ * @param wrapStream true if we want to wrap the inputstream based on filesystem specific criteria
* @return the InputStream to read from.
* @throws IOException IO error.
*/
@PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
- public abstract SeekableDataInputStream openSeekable(StoragePath path, int bufferSize) throws IOException;
+ public abstract SeekableDataInputStream openSeekable(StoragePath path, int bufferSize, boolean wrapStream) throws IOException;
/**
* Appends to an existing file (optional operation).
@@ -392,12 +393,13 @@ public abstract class HoodieStorage implements Closeable {
* Opens an SeekableDataInputStream at the indicated path with seeks supported.
*
* @param path the file to open.
+ * @param wrapStream true if we want to wrap the inputstream based on filesystem specific criteria
* @return the InputStream to read from.
* @throws IOException IO error.
*/
@PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
- public SeekableDataInputStream openSeekable(StoragePath path) throws IOException {
- return openSeekable(path, getDefaultBlockSize(path));
+ public SeekableDataInputStream openSeekable(StoragePath path, boolean wrapStream) throws IOException {
+ return openSeekable(path, getDefaultBlockSize(path), wrapStream);
}
/**
diff --git a/hudi-io/src/main/java/org/apache/hudi/storage/StorageConfiguration.java b/hudi-io/src/main/java/org/apache/hudi/storage/StorageConfiguration.java
index ac586fc6f72..15f0333fd5b 100644
--- a/hudi-io/src/main/java/org/apache/hudi/storage/StorageConfiguration.java
+++ b/hudi-io/src/main/java/org/apache/hudi/storage/StorageConfiguration.java
@@ -63,6 +63,13 @@ public abstract class StorageConfiguration<T> implements Serializable {
*/
public abstract Option<String> getString(String key);
+ /**
+ * Gets an inline version of this storage configuration
+ *
+ * @return copy of this storage configuration that is inline
+ */
+ public abstract StorageConfiguration<T> getInline();
+
/**
* @param clazz class of U, which is assignable from T.
* @param <U> type to return.
diff --git a/hudi-io/src/main/java/org/apache/hudi/storage/StoragePath.java b/hudi-io/src/main/java/org/apache/hudi/storage/StoragePath.java
index 24bf77e76ad..2a24978f084 100644
--- a/hudi-io/src/main/java/org/apache/hudi/storage/StoragePath.java
+++ b/hudi-io/src/main/java/org/apache/hudi/storage/StoragePath.java
@@ -235,6 +235,13 @@ public class StoragePath implements Comparable<StoragePath>, Serializable {
return new StoragePath(newUri);
}
+ @PublicAPIMethod(maturity = ApiMaturityLevel.EVOLVING)
+ public String getFileExtension() {
+ String fileName = getName();
+ int dotIndex = fileName.lastIndexOf('.');
+ return dotIndex == -1 ? "" : fileName.substring(dotIndex);
+ }
+
@Override
public String toString() {
// This value could be overwritten concurrently and that's okay, since
diff --git a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/inline/InLineFSUtils.java b/hudi-io/src/main/java/org/apache/hudi/storage/inline/InLineFSUtils.java
similarity index 65%
rename from hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/inline/InLineFSUtils.java
rename to hudi-io/src/main/java/org/apache/hudi/storage/inline/InLineFSUtils.java
index 6c6cb7323e4..97b8de50050 100644
--- a/hudi-hadoop-common/src/main/java/org/apache/hudi/hadoop/fs/inline/InLineFSUtils.java
+++ b/hudi-io/src/main/java/org/apache/hudi/storage/inline/InLineFSUtils.java
@@ -17,28 +17,22 @@
* under the License.
*/
-package org.apache.hudi.hadoop.fs.inline;
+package org.apache.hudi.storage.inline;
import org.apache.hudi.storage.StoragePath;
-import org.apache.hadoop.fs.Path;
-
import java.io.File;
import static org.apache.hudi.common.util.ValidationUtils.checkArgument;
-/**
- * Utils to parse InLineFileSystem paths.
- * Inline FS format:
- * "inlinefs://<path_to_outer_file>/<outer_file_scheme>/?start_offset=start_offset>&length=<length>"
- * Eg: "inlinefs://<path_to_outer_file>/s3a/?start_offset=20&length=40"
- */
public class InLineFSUtils {
- private static final String START_OFFSET_STR = "start_offset";
- private static final String LENGTH_STR = "length";
- private static final String SCHEME_SEPARATOR = "" + StoragePath.COLON_CHAR;
- private static final String EQUALS_STR = "=";
- private static final String LOCAL_FILESYSTEM_SCHEME = "file";
+
+ public static final String SCHEME = "inlinefs";
+ protected static final String START_OFFSET_STR = "start_offset";
+ protected static final String LENGTH_STR = "length";
+ protected static final String SCHEME_SEPARATOR = "" + StoragePath.COLON_CHAR;
+ protected static final String EQUALS_STR = "=";
+ protected static final String LOCAL_FILESYSTEM_SCHEME = "file";
/**
* Get the InlineFS Path for a given schema and its Path.
@@ -59,42 +53,13 @@ public class InLineFSUtils {
long inLineLength) {
final String subPath = new File(outerPath.toString().substring(outerPath.toString().indexOf(":") + 1)).getPath();
return new StoragePath(
- InLineFileSystem.SCHEME + SCHEME_SEPARATOR
+ SCHEME + SCHEME_SEPARATOR
+ StoragePath.SEPARATOR + subPath + StoragePath.SEPARATOR + origScheme
+ StoragePath.SEPARATOR + "?" + START_OFFSET_STR + EQUALS_STR + inLineStartOffset
+ "&" + LENGTH_STR + EQUALS_STR + inLineLength
);
}
- /**
- * InlineFS Path format:
- * "inlinefs://path/to/outer/file/outer_file_scheme/?start_offset=start_offset>&length=<length>"
- * <p>
- * Outer File Path format:
- * "outer_file_scheme://path/to/outer/file"
- * <p>
- * Example
- * Input: "inlinefs://file1/s3a/?start_offset=20&length=40".
- * Output: "s3a://file1"
- *
- * @param inlineFSPath InLineFS Path to get the outer file Path
- * @return Outer file Path from the InLineFS Path
- */
- public static Path getOuterFilePathFromInlinePath(Path inlineFSPath) {
- assertInlineFSPath(inlineFSPath);
-
- final String outerFileScheme = inlineFSPath.getParent().getName();
- final Path basePath = inlineFSPath.getParent().getParent();
- checkArgument(basePath.toString().contains(SCHEME_SEPARATOR),
- "Invalid InLineFS path: " + inlineFSPath);
-
- final String pathExceptScheme = basePath.toString().substring(basePath.toString().indexOf(SCHEME_SEPARATOR) + 1);
- final String fullPath = outerFileScheme + SCHEME_SEPARATOR
- + (outerFileScheme.equals(LOCAL_FILESYSTEM_SCHEME) ? StoragePath.SEPARATOR : "")
- + pathExceptScheme;
- return new Path(fullPath);
- }
-
public static StoragePath getOuterFilePathFromInlinePath(StoragePath inlineFSPath) {
assertInlineFSPath(inlineFSPath);
@@ -136,13 +101,8 @@ public class InLineFSUtils {
return Long.parseLong(slices[slices.length - 1]);
}
- private static void assertInlineFSPath(Path inlinePath) {
- String scheme = inlinePath.toUri().getScheme();
- checkArgument(InLineFileSystem.SCHEME.equals(scheme));
- }
-
private static void assertInlineFSPath(StoragePath inlinePath) {
String scheme = inlinePath.toUri().getScheme();
- checkArgument(InLineFileSystem.SCHEME.equals(scheme));
+ checkArgument(SCHEME.equals(scheme));
}
}
diff --git a/hudi-common/src/test/java/org/apache/hudi/common/testutils/NetworkTestUtils.java b/hudi-io/src/test/java/org/apache/hudi/common/testutils/NetworkTestUtils.java
similarity index 100%
rename from hudi-common/src/test/java/org/apache/hudi/common/testutils/NetworkTestUtils.java
rename to hudi-io/src/test/java/org/apache/hudi/common/testutils/NetworkTestUtils.java
diff --git a/hudi-io/src/test/java/org/apache/hudi/io/storage/TestHoodieStorageBase.java b/hudi-io/src/test/java/org/apache/hudi/io/storage/TestHoodieStorageBase.java
index 0e40b562f66..cdc8d6f6746 100644
--- a/hudi-io/src/test/java/org/apache/hudi/io/storage/TestHoodieStorageBase.java
+++ b/hudi-io/src/test/java/org/apache/hudi/io/storage/TestHoodieStorageBase.java
@@ -163,11 +163,11 @@ public abstract class TestHoodieStorageBase {
stream.flush();
}
- try (SeekableDataInputStream seekableStream = storage.openSeekable(path)) {
+ try (SeekableDataInputStream seekableStream = storage.openSeekable(path, true)) {
validateSeekableDataInputStream(seekableStream, data);
}
- try (SeekableDataInputStream seekableStream = storage.openSeekable(path, 2)) {
+ try (SeekableDataInputStream seekableStream = storage.openSeekable(path, 2, true)) {
validateSeekableDataInputStream(seekableStream, data);
}
}
diff --git a/hudi-kafka-connect/pom.xml b/hudi-kafka-connect/pom.xml
index ceaffe936ad..40033448697 100644
--- a/hudi-kafka-connect/pom.xml
+++ b/hudi-kafka-connect/pom.xml
@@ -210,6 +210,14 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git a/hudi-spark-datasource/hudi-spark-common/pom.xml b/hudi-spark-datasource/hudi-spark-common/pom.xml
index 7a0930e1340..774acf52327 100644
--- a/hudi-spark-datasource/hudi-spark-common/pom.xml
+++ b/hudi-spark-datasource/hudi-spark-common/pom.xml
@@ -176,6 +176,16 @@
<artifactId>hudi-common</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-hive-sync</artifactId>
@@ -259,6 +269,22 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
</dependencies>
diff --git a/hudi-spark-datasource/hudi-spark/pom.xml b/hudi-spark-datasource/hudi-spark/pom.xml
index 5072f445db6..8ebb11a2386 100644
--- a/hudi-spark-datasource/hudi-spark/pom.xml
+++ b/hudi-spark-datasource/hudi-spark/pom.xml
@@ -201,6 +201,16 @@
<artifactId>hudi-common</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-io</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-hadoop-mr</artifactId>
@@ -458,6 +468,22 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-io</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-java-client</artifactId>
diff --git a/hudi-spark-datasource/hudi-spark2/pom.xml b/hudi-spark-datasource/hudi-spark2/pom.xml
index 57c849026c6..7c435d42adc 100644
--- a/hudi-spark-datasource/hudi-spark2/pom.xml
+++ b/hudi-spark-datasource/hudi-spark2/pom.xml
@@ -183,6 +183,11 @@
<artifactId>hudi-common</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-spark-common_${scala.binary.version}</artifactId>
@@ -236,6 +241,14 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-spark-common_${scala.binary.version}</artifactId>
diff --git a/hudi-spark-datasource/hudi-spark3-common/pom.xml b/hudi-spark-datasource/hudi-spark3-common/pom.xml
index 83619b3f19a..524dd057fa4 100644
--- a/hudi-spark-datasource/hudi-spark3-common/pom.xml
+++ b/hudi-spark-datasource/hudi-spark3-common/pom.xml
@@ -225,6 +225,14 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-spark-common_${scala.binary.version}</artifactId>
diff --git a/hudi-spark-datasource/hudi-spark3.0.x/pom.xml b/hudi-spark-datasource/hudi-spark3.0.x/pom.xml
index 8418ac2f0e5..d7c7a47ec7e 100644
--- a/hudi-spark-datasource/hudi-spark3.0.x/pom.xml
+++ b/hudi-spark-datasource/hudi-spark3.0.x/pom.xml
@@ -256,6 +256,14 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
diff --git a/hudi-spark-datasource/hudi-spark3.1.x/pom.xml b/hudi-spark-datasource/hudi-spark3.1.x/pom.xml
index 0c0609d4510..5b351489e77 100644
--- a/hudi-spark-datasource/hudi-spark3.1.x/pom.xml
+++ b/hudi-spark-datasource/hudi-spark3.1.x/pom.xml
@@ -256,6 +256,14 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
diff --git a/hudi-spark-datasource/hudi-spark3.2.x/pom.xml b/hudi-spark-datasource/hudi-spark3.2.x/pom.xml
index 0078178422e..d463fd99453 100644
--- a/hudi-spark-datasource/hudi-spark3.2.x/pom.xml
+++ b/hudi-spark-datasource/hudi-spark3.2.x/pom.xml
@@ -301,6 +301,15 @@
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-spark-common_${scala.binary.version}</artifactId>
diff --git a/hudi-spark-datasource/hudi-spark3.2plus-common/pom.xml b/hudi-spark-datasource/hudi-spark3.2plus-common/pom.xml
index e9e90c57a2f..708c59805a6 100644
--- a/hudi-spark-datasource/hudi-spark3.2plus-common/pom.xml
+++ b/hudi-spark-datasource/hudi-spark3.2plus-common/pom.xml
@@ -210,6 +210,14 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-spark-common_${scala.binary.version}</artifactId>
diff --git a/hudi-spark-datasource/hudi-spark3.3.x/pom.xml b/hudi-spark-datasource/hudi-spark3.3.x/pom.xml
index ae3477f2e49..0b39aa299c9 100644
--- a/hudi-spark-datasource/hudi-spark3.3.x/pom.xml
+++ b/hudi-spark-datasource/hudi-spark3.3.x/pom.xml
@@ -291,6 +291,22 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
diff --git a/hudi-spark-datasource/hudi-spark3.4.x/pom.xml b/hudi-spark-datasource/hudi-spark3.4.x/pom.xml
index 92f63cacb96..9ecdb92559d 100644
--- a/hudi-spark-datasource/hudi-spark3.4.x/pom.xml
+++ b/hudi-spark-datasource/hudi-spark3.4.x/pom.xml
@@ -292,6 +292,15 @@
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-spark-common_${scala.binary.version}</artifactId>
diff --git a/hudi-spark-datasource/hudi-spark3.5.x/pom.xml b/hudi-spark-datasource/hudi-spark3.5.x/pom.xml
index edd3f911969..9e24f7c8fbd 100644
--- a/hudi-spark-datasource/hudi-spark3.5.x/pom.xml
+++ b/hudi-spark-datasource/hudi-spark3.5.x/pom.xml
@@ -292,6 +292,15 @@
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-spark-common_${scala.binary.version}</artifactId>
diff --git a/hudi-sync/hudi-hive-sync/pom.xml b/hudi-sync/hudi-hive-sync/pom.xml
index 69aa590bf2d..f535642ea95 100644
--- a/hudi-sync/hudi-hive-sync/pom.xml
+++ b/hudi-sync/hudi-hive-sync/pom.xml
@@ -49,6 +49,16 @@
<artifactId>hudi-common</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-io</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-hadoop-mr</artifactId>
@@ -120,7 +130,22 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
-
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-io</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-tests-common</artifactId>
diff --git a/hudi-sync/hudi-sync-common/pom.xml b/hudi-sync/hudi-sync-common/pom.xml
index 82d4152ed23..385b2edbb19 100644
--- a/hudi-sync/hudi-sync-common/pom.xml
+++ b/hudi-sync/hudi-sync-common/pom.xml
@@ -44,7 +44,11 @@
<artifactId>hudi-common</artifactId>
<version>${project.version}</version>
</dependency>
-
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<!-- Kryo -->
<dependency>
<groupId>com.esotericsoftware</groupId>
@@ -72,6 +76,14 @@
<version>${project.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
diff --git a/hudi-timeline-service/pom.xml b/hudi-timeline-service/pom.xml
index 4086eb98401..56a1890b486 100644
--- a/hudi-timeline-service/pom.xml
+++ b/hudi-timeline-service/pom.xml
@@ -78,6 +78,11 @@
<artifactId>hudi-common</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<!-- Kryo -->
<dependency>
@@ -160,6 +165,14 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git a/hudi-utilities/pom.xml b/hudi-utilities/pom.xml
index 60ab26b4f0b..ad4806655c4 100644
--- a/hudi-utilities/pom.xml
+++ b/hudi-utilities/pom.xml
@@ -168,6 +168,16 @@
<artifactId>hudi-common</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-io</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
@@ -442,6 +452,22 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-io</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hudi</groupId>
+ <artifactId>hudi-hadoop-common</artifactId>
+ <version>${project.version}</version>
+ <classifier>tests</classifier>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.hudi</groupId>
<artifactId>hudi-hive-sync</artifactId>