This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 65cf7e8ede7d test: improve meta sync coverage (#19539)
65cf7e8ede7d is described below
commit 65cf7e8ede7d1a08f3b3f13b4a45c6229043cc3e
Author: Danny Chan <[email protected]>
AuthorDate: Fri Aug 7 17:16:05 2026 +0800
test: improve meta sync coverage (#19539)
* test: improve meta sync coverage
---
.../hive/TestHoodieHiveSyncClientOperations.java | 226 +++++++++++++++++++++
.../hive/ddl/TestHiveQueryDDLExecutorFailures.java | 85 ++++++++
.../hive/ddl/TestJDBCBasedMetadataOperator.java | 24 +++
.../TestHiveSyncGlobalCommitToolUnit.java | 90 ++++++++
...iveMetastoreBasedLockProviderConfiguration.java | 68 +++++++
.../sync/common/TestHoodieMetaSyncOperations.java | 72 +++++++
6 files changed, 565 insertions(+)
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHoodieHiveSyncClientOperations.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHoodieHiveSyncClientOperations.java
new file mode 100644
index 000000000000..c380cbb535e7
--- /dev/null
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/TestHoodieHiveSyncClientOperations.java
@@ -0,0 +1,226 @@
+/*
+ * 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.hive;
+
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.hive.ddl.DDLExecutor;
+import org.apache.hudi.hive.ddl.JDBCBasedMetadataOperator;
+import org.apache.hudi.sync.common.model.FieldSchema;
+import org.apache.hudi.sync.common.model.Partition;
+
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
+import org.apache.hadoop.hive.metastore.api.SerDeInfo;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.thrift.TApplicationException;
+import org.apache.thrift.TException;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.InOrder;
+
+import java.lang.reflect.Field;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+
+import static
org.apache.hudi.hadoop.utils.HoodieHiveUtils.GLOBALLY_CONSISTENT_READ_TIMESTAMP;
+import static
org.apache.hudi.sync.common.HoodieMetaSyncOperations.HOODIE_LAST_COMMIT_COMPLETION_TIME_SYNC;
+import static
org.apache.hudi.sync.common.HoodieMetaSyncOperations.HOODIE_LAST_COMMIT_TIME_SYNC;
+import static
org.apache.hudi.sync.common.HoodieSyncConfig.META_SYNC_BASE_FILE_FORMAT;
+import static org.apache.hudi.sync.common.HoodieSyncConfig.META_SYNC_BASE_PATH;
+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.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class TestHoodieHiveSyncClientOperations {
+
+ @Test
+ void thriftMismatchActivatesJdbcMetadataFallback() throws Exception {
+ TestFixture fixture = newFixture();
+ TApplicationException incompatible = new TApplicationException("unknown
get_table method");
+ when(fixture.metaStoreClient.tableExists("test_db", "table"))
+ .thenThrow(new TException(incompatible));
+ when(fixture.jdbcMetadataOperator.tableExists("table")).thenReturn(true);
+
when(fixture.jdbcMetadataOperator.databaseExists("test_db")).thenReturn(true);
+ when(fixture.jdbcMetadataOperator.getTableProperty("table",
HOODIE_LAST_COMMIT_TIME_SYNC))
+ .thenReturn(Option.of("100"));
+ when(fixture.jdbcMetadataOperator.getTableProperty("table",
HOODIE_LAST_COMMIT_COMPLETION_TIME_SYNC))
+ .thenReturn(Option.of("101"));
+ when(fixture.jdbcMetadataOperator.getTableProperty("table",
GLOBALLY_CONSISTENT_READ_TIMESTAMP))
+ .thenReturn(Option.of("099"));
+
when(fixture.jdbcMetadataOperator.getTableLocation("table")).thenReturn("/warehouse/table");
+ List<FieldSchema> fields = Collections.singletonList(new FieldSchema("id",
"string", Option.empty()));
+
when(fixture.jdbcMetadataOperator.getFieldSchemas("table")).thenReturn(fields);
+ List<Partition> partitions = Collections.singletonList(
+ new Partition(Collections.singletonList("2026-08-06"),
"/base/datestr=2026-08-06"));
+ when(fixture.jdbcMetadataOperator.getAllPartitions("table",
"/base")).thenReturn(partitions);
+
+ assertTrue(fixture.syncClient.tableExists("table"));
+ assertTrue(fixture.syncClient.updateTableProperties("table",
Collections.singletonMap("owner", "hudi")));
+ assertTrue(fixture.syncClient.updateSerdeProperties("table",
+ new HashMap<>(Collections.singletonMap("compression", "none")),
false));
+ assertEquals(partitions, fixture.syncClient.getAllPartitions("table"));
+ assertEquals(partitions,
+ fixture.syncClient.getPartitionsFromList("table",
Collections.singletonList("datestr=2026-08-06")));
+ assertTrue(fixture.syncClient.databaseExists("test_db"));
+ assertEquals("100",
fixture.syncClient.getLastCommitTimeSynced("table").get());
+ assertEquals("101",
fixture.syncClient.getLastCommitCompletionTimeSynced("table").get());
+ assertEquals("099",
fixture.syncClient.getLastReplicatedTime("table").get());
+ assertEquals(fields, fixture.syncClient.getMetastoreFieldSchemas("table"));
+ assertEquals("/warehouse/table",
fixture.syncClient.getTableLocation("table"));
+
+ fixture.syncClient.deleteLastReplicatedTimeStamp("table");
+ fixture.syncClient.createOrReplaceTable("table", null, "input", "output",
"serde",
+ Collections.emptyMap(), Collections.emptyMap());
+
+ ArgumentCaptor<String> temporaryTableNameCaptor =
ArgumentCaptor.forClass(String.class);
+ InOrder replacementOrder = inOrder(fixture.ddlExecutor,
fixture.jdbcMetadataOperator);
+
replacementOrder.verify(fixture.ddlExecutor).createTable(temporaryTableNameCaptor.capture(),
+ isNull(), eq("input"), eq("output"), eq("serde"),
eq(Collections.emptyMap()), eq(Collections.emptyMap()));
+ String temporaryTableName = temporaryTableNameCaptor.getValue();
+ replacementOrder.verify(fixture.jdbcMetadataOperator).dropTable("table");
+
replacementOrder.verify(fixture.jdbcMetadataOperator).renameTable(temporaryTableName,
"table");
+
+ verify(fixture.jdbcMetadataOperator).unsetTableProperty("table",
GLOBALLY_CONSISTENT_READ_TIMESTAMP);
+ verify(fixture.jdbcMetadataOperator).setStorageFormat(eq("table"),
anyString(), anyString(), anyString(), any());
+ }
+
+ @Test
+ void metastoreFailuresAreWrappedWhenNoFallbackIsConfigured() throws
Exception {
+ TestFixture fixture = newFixtureWithoutJdbc();
+
+ assertFalse(fixture.syncClient.updateTableProperties("table",
Collections.emptyMap()));
+ assertFalse(fixture.syncClient.updateSerdeProperties("table",
Collections.emptyMap(), false));
+
+ when(fixture.metaStoreClient.tableExists("test_db",
"missing")).thenReturn(false);
+ assertThrows(IllegalArgumentException.class, () ->
fixture.syncClient.getMetastoreSchema("missing"));
+
+ when(fixture.metaStoreClient.tableExists("test_db",
"table")).thenThrow(new TException("unavailable"));
+ assertThrows(HoodieHiveSyncException.class, () ->
fixture.syncClient.tableExists("table"));
+
+ when(fixture.metaStoreClient.getDatabase("test_db")).thenThrow(new
TException("unavailable"));
+ assertThrows(HoodieHiveSyncException.class, () ->
fixture.syncClient.databaseExists("test_db"));
+
+ when(fixture.metaStoreClient.getTable("test_db", "absent"))
+ .thenThrow(new NoSuchObjectException("absent"));
+
assertFalse(fixture.syncClient.getLastReplicatedTime("absent").isPresent());
+ fixture.syncClient.deleteLastReplicatedTimeStamp("absent");
+
+ when(fixture.metaStoreClient.getTable("test_db", "table")).thenThrow(new
TException("unavailable"));
+ assertThrows(HoodieHiveSyncException.class, () ->
fixture.syncClient.getLastCommitTimeSynced("table"));
+ assertThrows(HoodieHiveSyncException.class,
+ () -> fixture.syncClient.getLastCommitCompletionTimeSynced("table"));
+ assertThrows(HoodieHiveSyncException.class, () ->
fixture.syncClient.getLastReplicatedTime("table"));
+ assertThrows(HoodieHiveSyncException.class, () ->
fixture.syncClient.deleteLastReplicatedTimeStamp("table"));
+ assertThrows(HoodieHiveSyncException.class,
+ () -> fixture.syncClient.updateTableProperties("table",
Collections.singletonMap("key", "value")));
+ assertThrows(HoodieHiveSyncException.class,
+ () -> fixture.syncClient.updateSerdeProperties("table", new
HashMap<>(Collections.singletonMap("key", "value")), false));
+ assertThrows(HoodieHiveSyncException.class, () ->
fixture.syncClient.getTableLocation("table"));
+ assertThrows(HoodieHiveSyncException.class, () ->
fixture.syncClient.updateHoodieWriterVersion("table"));
+
+ when(fixture.metaStoreClient.getSchema("test_db", "table")).thenThrow(new
TException("unavailable"));
+ assertThrows(HoodieHiveSyncException.class, () ->
fixture.syncClient.getMetastoreFieldSchemas("table"));
+ org.mockito.Mockito.doThrow(new TException("unavailable"))
+ .when(fixture.metaStoreClient).dropTable("test_db", "table");
+ assertThrows(HoodieHiveSyncException.class, () ->
fixture.syncClient.dropTable("table"));
+ }
+
+ @Test
+ void metastoreCreateReplaceAndSerdeAlterPathsUpdateMetadata() throws
Exception {
+ TestFixture fixture = newFixtureWithoutJdbc();
+ when(fixture.metaStoreClient.tableExists("test_db",
"table")).thenReturn(true);
+ Table temporaryTable = new Table();
+ when(fixture.metaStoreClient.getTable(eq("test_db"),
anyString())).thenReturn(temporaryTable);
+
+ fixture.syncClient.createOrReplaceTable("table", null, "input", "output",
"serde",
+ Collections.emptyMap(), Collections.emptyMap());
+
+ ArgumentCaptor<String> temporaryTableNameCaptor =
ArgumentCaptor.forClass(String.class);
+ InOrder replacementOrder = inOrder(fixture.ddlExecutor,
fixture.metaStoreClient);
+
replacementOrder.verify(fixture.ddlExecutor).createTable(temporaryTableNameCaptor.capture(),
+ isNull(), eq("input"), eq("output"), eq("serde"),
eq(Collections.emptyMap()), eq(Collections.emptyMap()));
+ String temporaryTableName = temporaryTableNameCaptor.getValue();
+ replacementOrder.verify(fixture.metaStoreClient).dropTable("test_db",
"table");
+ replacementOrder.verify(fixture.metaStoreClient).alter_table("test_db",
temporaryTableName, temporaryTable);
+
+ Table serdeTable = new Table();
+ StorageDescriptor storageDescriptor = new StorageDescriptor();
+ storageDescriptor.setInputFormat("old-input");
+ storageDescriptor.setOutputFormat("old-output");
+ storageDescriptor.setSerdeInfo((SerDeInfo) null);
+ serdeTable.setSd(storageDescriptor);
+ when(fixture.metaStoreClient.getTable("test_db",
"serde_table")).thenReturn(serdeTable);
+
+ assertTrue(fixture.syncClient.updateSerdeProperties("serde_table",
+ new HashMap<>(Collections.singletonMap("compression", "none")),
false));
+
assertTrue(storageDescriptor.getSerdeInfo().getParameters().containsKey("serialization.format"));
+ assertFalse(fixture.syncClient.updateTableComments("table",
Collections.emptyList(), Collections.emptyList()));
+ }
+
+ private static TestFixture newFixture() throws Exception {
+ TestFixture fixture = newFixtureWithoutJdbc();
+ fixture.jdbcMetadataOperator = mock(JDBCBasedMetadataOperator.class);
+ setField(fixture.syncClient, "jdbcMetadataOperator",
fixture.jdbcMetadataOperator);
+ return fixture;
+ }
+
+ private static TestFixture newFixtureWithoutJdbc() throws Exception {
+ TestFixture fixture = new TestFixture();
+ fixture.syncClient = mock(HoodieHiveSyncClient.class, CALLS_REAL_METHODS);
+ fixture.config = mock(HiveSyncConfig.class);
+ fixture.metaStoreClient = mock(IMetaStoreClient.class);
+ fixture.ddlExecutor = mock(DDLExecutor.class);
+ when(fixture.config.getString(META_SYNC_BASE_PATH)).thenReturn("/base");
+
when(fixture.config.getStringOrDefault(META_SYNC_BASE_FILE_FORMAT)).thenReturn("PARQUET");
+
when(fixture.ddlExecutor.supportsUpdatingPartitionColumnComments()).thenReturn(true);
+ setField(fixture.syncClient, "config", fixture.config);
+ setField(fixture.syncClient, "databaseName", "test_db");
+ setField(fixture.syncClient, "client", fixture.metaStoreClient);
+ setField(fixture.syncClient, "ddlExecutor", fixture.ddlExecutor);
+ setField(fixture.syncClient, "initialTableByName", new HashMap<String,
Table>());
+ return fixture;
+ }
+
+ private static void setField(Object target, String name, Object value)
throws Exception {
+ Field field = HoodieHiveSyncClient.class.getDeclaredField(name);
+ field.setAccessible(true);
+ field.set(target, value);
+ }
+
+ private static class TestFixture {
+ private HoodieHiveSyncClient syncClient;
+ private HiveSyncConfig config;
+ private IMetaStoreClient metaStoreClient;
+ private DDLExecutor ddlExecutor;
+ private JDBCBasedMetadataOperator jdbcMetadataOperator;
+ }
+}
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/ddl/TestHiveQueryDDLExecutorFailures.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/ddl/TestHiveQueryDDLExecutorFailures.java
new file mode 100644
index 000000000000..c9d353ac7718
--- /dev/null
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/ddl/TestHiveQueryDDLExecutorFailures.java
@@ -0,0 +1,85 @@
+/*
+ * 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.hive.ddl;
+
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.hive.HiveSyncConfig;
+import org.apache.hudi.hive.HoodieHiveSyncException;
+import org.apache.hudi.sync.common.model.PartitionValueExtractor;
+
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.apache.thrift.TException;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.util.Collections;
+
+import static
org.apache.hudi.sync.common.HoodieSyncConfig.META_SYNC_DATABASE_NAME;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class TestHiveQueryDDLExecutorFailures {
+
+ @Test
+ void emptyAndMetastoreFailurePathsAreHandled() throws Exception {
+ HiveQueryDDLExecutor executor = mock(HiveQueryDDLExecutor.class,
CALLS_REAL_METHODS);
+ IMetaStoreClient metaStoreClient = mock(IMetaStoreClient.class);
+ HiveSyncConfig config = mock(HiveSyncConfig.class);
+ PartitionValueExtractor partitionValueExtractor =
mock(PartitionValueExtractor.class);
+ setField(executor, "driverPool", Option.empty());
+ setField(executor, "metaStoreClient", metaStoreClient);
+ setField(executor, "databaseName", "test_db");
+ setField(executor, "config", config);
+ setField(executor, "partitionValueExtractor", partitionValueExtractor);
+
+ assertDoesNotThrow(() -> executor.runSQLs(Collections.emptyList()));
+ assertDoesNotThrow(() -> executor.dropPartitionsToTable("table",
Collections.emptyList()));
+
+ when(metaStoreClient.getTable(anyString(), anyString())).thenThrow(new
TException("unavailable"));
+
when(config.getStringOrDefault(META_SYNC_DATABASE_NAME)).thenReturn("test_db");
+ when(partitionValueExtractor.extractPartitionValuesInPath(anyString()))
+ .thenReturn(Collections.singletonList("2026-08-06"));
+ when(metaStoreClient.getPartition(anyString(), anyString(), anyList()))
+ .thenThrow(new TException("unavailable"));
+ assertThrows(HoodieHiveSyncException.class, () ->
executor.getTableSchema("table"));
+ assertThrows(HoodieHiveSyncException.class,
+ () -> executor.dropPartitionsToTable("table",
Collections.singletonList("datestr=2026-08-06")));
+ }
+
+ private static void setField(Object target, String name, Object value)
throws Exception {
+ Class<?> type = target.getClass();
+ while (type != null) {
+ try {
+ Field field = type.getDeclaredField(name);
+ field.setAccessible(true);
+ field.set(target, value);
+ return;
+ } catch (NoSuchFieldException e) {
+ type = type.getSuperclass();
+ }
+ }
+ throw new NoSuchFieldException(name);
+ }
+}
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/ddl/TestJDBCBasedMetadataOperator.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/ddl/TestJDBCBasedMetadataOperator.java
index 65ca35a8610d..0c1e0620be1d 100644
---
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/ddl/TestJDBCBasedMetadataOperator.java
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/ddl/TestJDBCBasedMetadataOperator.java
@@ -19,6 +19,7 @@
package org.apache.hudi.hive.ddl;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.hive.HoodieHiveSyncException;
import org.apache.hudi.sync.common.model.FieldSchema;
import org.apache.hudi.sync.common.model.Partition;
@@ -34,6 +35,7 @@ import java.util.Map;
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.anyString;
import static org.mockito.Mockito.mock;
@@ -101,6 +103,19 @@ class TestJDBCBasedMetadataOperator {
assertEquals("name", fields.get(1).getName());
}
+ @Test
+ void testGetFieldSchemasHandlesMissingType() throws Exception {
+ when(mockStatement.executeQuery(anyString())).thenReturn(mockResultSet);
+ when(mockResultSet.next()).thenReturn(true, false);
+ when(mockResultSet.getString(1)).thenReturn("mystery");
+ when(mockResultSet.getString(2)).thenReturn(null);
+
+ List<FieldSchema> fields = operator.getFieldSchemas("my_table");
+
+ assertEquals(1, fields.size());
+ assertEquals("", fields.get(0).getType());
+ }
+
@Test
void testGetTableProperty() throws Exception {
ResultSetMetaData metaData = mock(ResultSetMetaData.class);
@@ -165,6 +180,15 @@ class TestJDBCBasedMetadataOperator {
assertEquals("s3a://bucket/warehouse/table", location);
}
+ @Test
+ void testGetTableLocationFailsWhenDescribeHasNoLocation() throws Exception {
+ when(mockStatement.executeQuery(anyString())).thenReturn(mockResultSet);
+ when(mockResultSet.next()).thenReturn(true, false);
+ when(mockResultSet.getString(1)).thenReturn("# Detailed Table
Information");
+
+ assertThrows(HoodieHiveSyncException.class, () ->
operator.getTableLocation("my_table"));
+ }
+
@Test
void testDropTable() throws Exception {
when(mockStatement.execute(anyString())).thenReturn(true);
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/replication/TestHiveSyncGlobalCommitToolUnit.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/replication/TestHiveSyncGlobalCommitToolUnit.java
new file mode 100644
index 000000000000..48ce525f1a3f
--- /dev/null
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/replication/TestHiveSyncGlobalCommitToolUnit.java
@@ -0,0 +1,90 @@
+/*
+ * 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.hive.replication;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.io.OutputStream;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Arrays;
+import java.util.Properties;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.mockito.Mockito.CALLS_REAL_METHODS;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class TestHiveSyncGlobalCommitToolUnit {
+
+ @TempDir
+ Path tempDir;
+
+ @Test
+ void commitDetectsDivergentReplicationStateAndClosesBothSides() throws
Exception {
+ ReplicationStateSync local = mock(ReplicationStateSync.class);
+ ReplicationStateSync remote = mock(ReplicationStateSync.class);
+ when(local.getClusterId()).thenReturn("LOCALSYNC");
+ when(remote.getClusterId()).thenReturn("REMOTESYNC");
+ when(local.replicationStateIsInSync(remote)).thenReturn(false);
+
+ HiveSyncGlobalCommitTool tool = mock(HiveSyncGlobalCommitTool.class,
CALLS_REAL_METHODS);
+ setField(tool, "replicationStateSyncList", Arrays.asList(local, remote));
+
+ assertFalse(tool.commit());
+ tool.close();
+
+ verify(local).sync();
+ verify(remote).sync();
+ verify(local).close();
+ verify(remote).close();
+ }
+
+ @Test
+ void loadParamsReadsXmlConfiguration() throws Exception {
+ Path configFile = tempDir.resolve("global-sync.xml");
+ Properties properties = new Properties();
+ properties.setProperty("marker", "loaded");
+ try (OutputStream output = Files.newOutputStream(configFile)) {
+ properties.storeToXML(output, "test");
+ }
+
+ Method loadParams =
HiveSyncGlobalCommitTool.class.getDeclaredMethod("loadParams", String[].class);
+ loadParams.setAccessible(true);
+ HiveSyncGlobalCommitParams params = (HiveSyncGlobalCommitParams)
loadParams.invoke(null, (Object) new String[] {
+ "--config-xml-file", configFile.toString(), "--replicated-timestamp",
"100",
+ "--base-path", tempDir.resolve("table").toString()
+ });
+
+ assertEquals("loaded", params.loadedProps.getProperty("marker"));
+ assertEquals("100",
params.globalHiveSyncConfigParams.globallyReplicatedTimeStamp);
+ }
+
+ private static void setField(Object target, String name, Object value)
throws Exception {
+ Field field = HiveSyncGlobalCommitTool.class.getDeclaredField(name);
+ field.setAccessible(true);
+ field.set(target, value);
+ }
+}
diff --git
a/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/TestHiveMetastoreBasedLockProviderConfiguration.java
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/TestHiveMetastoreBasedLockProviderConfiguration.java
new file mode 100644
index 000000000000..5abf1acf81fb
--- /dev/null
+++
b/hudi-sync/hudi-hive-sync/src/test/java/org/apache/hudi/hive/transaction/lock/TestHiveMetastoreBasedLockProviderConfiguration.java
@@ -0,0 +1,68 @@
+/*
+ * 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.hive.transaction.lock;
+
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.metastore.IMetaStoreClient;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Method;
+
+import static
org.apache.hudi.common.config.LockConfiguration.HIVE_METASTORE_URI_PROP_KEY;
+import static
org.apache.hudi.common.config.LockConfiguration.LOCK_ACQUIRE_NUM_RETRIES_PROP_KEY;
+import static
org.apache.hudi.common.config.LockConfiguration.LOCK_ACQUIRE_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY;
+import static
org.apache.hudi.common.config.LockConfiguration.ZK_CONNECT_URL_PROP_KEY;
+import static org.apache.hudi.common.config.LockConfiguration.ZK_PORT_PROP_KEY;
+import static
org.apache.hudi.common.config.LockConfiguration.ZK_SESSION_TIMEOUT_MS_PROP_KEY;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.mock;
+
+class TestHiveMetastoreBasedLockProviderConfiguration extends
HiveMetastoreBasedLockProviderTestBase {
+
+ @Test
+ void appliesOptionalMetastoreAndZooKeeperSettings() throws Exception {
+ lockConfiguration.getConfig().setProperty(HIVE_METASTORE_URI_PROP_KEY,
"thrift://localhost:9083");
+
lockConfiguration.getConfig().setProperty(LOCK_ACQUIRE_NUM_RETRIES_PROP_KEY,
"4");
+
lockConfiguration.getConfig().setProperty(LOCK_ACQUIRE_RETRY_WAIT_TIME_IN_MILLIS_PROP_KEY,
"25");
+ lockConfiguration.getConfig().setProperty(ZK_CONNECT_URL_PROP_KEY,
"zk-a,zk-b");
+ lockConfiguration.getConfig().setProperty(ZK_PORT_PROP_KEY, "2182");
+ lockConfiguration.getConfig().setProperty(ZK_SESSION_TIMEOUT_MS_PROP_KEY,
"45000");
+
+ HiveMetastoreBasedLockProvider provider = new
HiveMetastoreBasedLockProvider(
+ lockConfiguration, mock(IMetaStoreClient.class));
+ try {
+ HiveConf hiveConf = new HiveConf();
+ Method method =
HiveMetastoreBasedLockProvider.class.getDeclaredMethod("setHiveLockConfs",
HiveConf.class);
+ method.setAccessible(true);
+ method.invoke(provider, hiveConf);
+
+ assertEquals("thrift://localhost:9083",
hiveConf.getVar(HiveConf.ConfVars.METASTOREURIS));
+ assertEquals("zk-a,zk-b", hiveConf.get("hive.zookeeper.quorum"));
+ assertEquals("2182", hiveConf.get("hive.zookeeper.client.port"));
+ assertEquals("45000", hiveConf.get("hive.zookeeper.session.timeout"));
+ assertEquals("4", hiveConf.get("hive.lock.numretries"));
+ assertEquals("25", hiveConf.get("hive.lock.sleep.between.retries"));
+ assertTrue(hiveConf.getBoolean("hive.support.concurrency", false));
+ } finally {
+ provider.close();
+ }
+ }
+}
diff --git
a/hudi-sync/hudi-sync-common/src/test/java/org/apache/hudi/sync/common/TestHoodieMetaSyncOperations.java
b/hudi-sync/hudi-sync-common/src/test/java/org/apache/hudi/sync/common/TestHoodieMetaSyncOperations.java
new file mode 100644
index 000000000000..2d5bb10e17df
--- /dev/null
+++
b/hudi-sync/hudi-sync-common/src/test/java/org/apache/hudi/sync/common/TestHoodieMetaSyncOperations.java
@@ -0,0 +1,72 @@
+/*
+ * 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.sync.common;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class TestHoodieMetaSyncOperations {
+
+ private final HoodieMetaSyncOperations operations = new
HoodieMetaSyncOperations() { };
+
+ @Test
+ void defaultOperationsRemainSafeForMinimalImplementations() {
+ assertDoesNotThrow(() -> operations.createTable("table", null, null, null,
null,
+ Collections.emptyMap(), Collections.emptyMap()));
+ assertDoesNotThrow(() -> operations.createOrReplaceTable("table", null,
null, null, null,
+ Collections.emptyMap(), Collections.emptyMap()));
+ assertFalse(operations.tableExists("table"));
+ assertDoesNotThrow(() -> operations.dropTable("table"));
+ assertDoesNotThrow(() -> operations.addPartitionsToTable("table",
Collections.emptyList()));
+ assertDoesNotThrow(() -> operations.updatePartitionsToTable("table",
Collections.emptyList()));
+ assertDoesNotThrow(() -> operations.touchPartitionsToTable("table",
Collections.emptyList()));
+ assertDoesNotThrow(() -> operations.dropPartitions("table",
Collections.emptyList()));
+ assertTrue(operations.getAllPartitions("table").isEmpty());
+ assertTrue(operations.getPartitionsFromList("table",
Collections.emptyList()).isEmpty());
+ assertFalse(operations.databaseExists("database"));
+ assertDoesNotThrow(() -> operations.createDatabase("database"));
+ assertTrue(operations.getMetastoreSchema("table").isEmpty());
+ assertNull(operations.getStorageSchema());
+ assertNull(operations.getStorageSchema(true));
+ assertDoesNotThrow(() -> operations.updateTableSchema("table", null,
null));
+ assertTrue(operations.getMetastoreFieldSchemas("table").isEmpty());
+ assertTrue(operations.getStorageFieldSchemas().isEmpty());
+ assertTrue(operations.getTableLocation("table").isEmpty());
+ assertFalse(operations.updateTableComments("table",
Collections.emptyList(), Collections.emptyList()));
+ assertFalse(operations.getLastCommitTimeSynced("table").isPresent());
+
assertFalse(operations.getLastCommitCompletionTimeSynced("table").isPresent());
+ assertDoesNotThrow(() -> operations.updateLastCommitTimeSynced("table"));
+ assertFalse(operations.updateTableProperties("table",
Collections.emptyMap()));
+ assertFalse(operations.updateSerdeProperties("table",
Collections.emptyMap(), false));
+ assertFalse(operations.getLastReplicatedTime("table").isPresent());
+ assertDoesNotThrow(() -> operations.updateLastReplicatedTimeStamp("table",
"001"));
+ assertDoesNotThrow(() ->
operations.deleteLastReplicatedTimeStamp("table"));
+ assertThrows(UnsupportedOperationException.class,
+ () -> operations.generatePushDownFilter(Collections.emptyList(),
Collections.emptyList()));
+ assertThrows(UnsupportedOperationException.class, () ->
operations.updateHoodieWriterVersion("table"));
+ }
+}
