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 dd52940c3493 feat(flink): add Hive sync metrics to the write
coordinator (#20148)
dd52940c3493 is described below
commit dd52940c349327fa51da6ffc8973a334a44b5c7f
Author: Shihuan Liu <[email protected]>
AuthorDate: Wed Sep 30 21:51:34 2026 -0700
feat(flink): add Hive sync metrics to the write coordinator (#20148)
* feat(flink): add Hive sync metrics to the write coordinator
The Flink write coordinator runs Hive sync on its own executor, which
only logs a failure. A table that has stopped syncing therefore leaves
no trace outside the JobManager log.
Register FlinkHiveSyncMetrics on the coordinator's metric group when
Hive sync is enabled, and record every sync in doSyncHive():
- hiveSyncSuccessCount / hiveSyncFailureCount (counters)
- hiveSyncDurationMs (histogram over the last 100 syncs, failures included)
- hiveSyncLastSuccessTimeMs (gauge, epoch millis of the last completed sync)
* Add tests for the Hive sync metrics
* Count a sync skipped by hive_sync.ignore_exceptions as a failure
* refactor(flink): simplify Hive sync exception handling
---------
Co-authored-by: danny0405 <[email protected]>
---
.../apache/hudi/metrics/FlinkHiveSyncMetrics.java | 88 +++++++++
.../hudi/sink/StreamWriteOperatorCoordinator.java | 18 ++
.../hudi/metrics/TestFlinkHiveSyncMetrics.java | 201 +++++++++++++++++++++
.../sink/TestStreamWriteOperatorCoordinator.java | 103 +++++++++++
.../java/org/apache/hudi/hive/HiveSyncTool.java | 8 +
5 files changed, 418 insertions(+)
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/metrics/FlinkHiveSyncMetrics.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/metrics/FlinkHiveSyncMetrics.java
new file mode 100644
index 000000000000..cc25e6080dce
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/metrics/FlinkHiveSyncMetrics.java
@@ -0,0 +1,88 @@
+/*
+ * 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.metrics;
+
+import com.codahale.metrics.SlidingWindowReservoir;
+import org.apache.flink.dropwizard.metrics.DropwizardHistogramWrapper;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.MetricGroup;
+import org.apache.flink.metrics.ThreadSafeSimpleCounter;
+
+/**
+ * Metrics for the Hive sync that {@link
org.apache.hudi.sink.StreamWriteOperatorCoordinator} runs
+ * on the JobManager after each commit.
+ *
+ * <p>A sync runs on the coordinator's Hive sync executor after a checkpoint
commit, and on its event
+ * executor after the commit that ends a bounded input, so each metric may be
updated from either
+ * thread.
+ */
+public class FlinkHiveSyncMetrics extends HoodieFlinkMetrics {
+ private static final int HISTOGRAM_WINDOW_SIZE = 100;
+
+ static final String HIVE_SYNC_SUCCESS_COUNT = "hiveSyncSuccessCount";
+ static final String HIVE_SYNC_FAILURE_COUNT = "hiveSyncFailureCount";
+ static final String HIVE_SYNC_DURATION_MS = "hiveSyncDurationMs";
+ static final String HIVE_SYNC_LAST_SUCCESS_TIME_MS =
"hiveSyncLastSuccessTimeMs";
+
+ /** Number of syncs that completed. */
+ private final Counter syncSuccessCount = new ThreadSafeSimpleCounter();
+
+ /**
+ * Number of syncs that threw, or that did nothing because the metastore
client could not be created
+ * and {@code hive_sync.ignore_exceptions} swallowed the error. The sync
executor only logs a
+ * failure, so without this counter a table that has stopped syncing leaves
no trace outside the
+ * JobManager log.
+ */
+ private final Counter syncFailureCount = new ThreadSafeSimpleCounter();
+
+ /**
+ * Duration of each sync in milliseconds, failed ones included: a sync that
fails after retrying
+ * an unreachable metastore is the slow case this is meant to show.
+ */
+ private final Histogram syncDurationMs;
+
+ /** Wall-clock time of the last completed sync in epoch milliseconds, or 0
until the first one. */
+ private volatile long lastSyncSuccessTimeMs = 0L;
+
+ public FlinkHiveSyncMetrics(MetricGroup metricGroup) {
+ super(metricGroup);
+ this.syncDurationMs = new DropwizardHistogramWrapper(
+ new com.codahale.metrics.Histogram(new
SlidingWindowReservoir(HISTOGRAM_WINDOW_SIZE)));
+ }
+
+ @Override
+ public void registerMetrics() {
+ metricGroup.counter(HIVE_SYNC_SUCCESS_COUNT, syncSuccessCount);
+ metricGroup.counter(HIVE_SYNC_FAILURE_COUNT, syncFailureCount);
+ metricGroup.histogram(HIVE_SYNC_DURATION_MS, syncDurationMs);
+ metricGroup.gauge(HIVE_SYNC_LAST_SUCCESS_TIME_MS, () ->
lastSyncSuccessTimeMs);
+ }
+
+ public void markSyncSucceeded(long durationMs) {
+ syncDurationMs.update(durationMs);
+ syncSuccessCount.inc();
+ lastSyncSuccessTimeMs = System.currentTimeMillis();
+ }
+
+ public void markSyncFailed(long durationMs) {
+ syncDurationMs.update(durationMs);
+ syncFailureCount.inc();
+ }
+}
diff --git
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java
index 1cdc59a8286a..8a0e2a16ae99 100644
---
a/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java
+++
b/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/StreamWriteOperatorCoordinator.java
@@ -25,6 +25,7 @@ import org.apache.hudi.common.model.WriteOperationType;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.timeline.HoodieTimeline;
import org.apache.hudi.common.util.CommitUtils;
+import org.apache.hudi.common.util.HoodieTimer;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.common.util.SerializationUtils;
import org.apache.hudi.common.util.ValidationUtils;
@@ -35,6 +36,7 @@ import org.apache.hudi.configuration.OptionsResolver;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.hive.HiveSyncTool;
+import org.apache.hudi.metrics.FlinkHiveSyncMetrics;
import org.apache.hudi.sink.common.AbstractStreamWriteFunction;
import org.apache.hudi.sink.event.Correspondent;
import org.apache.hudi.sink.event.WriteMetadataEvent;
@@ -190,6 +192,11 @@ public class StreamWriteOperatorCoordinator
*/
private HiveSyncContext hiveSyncContext;
+ /**
+ * Metrics for hive sync, registered on the coordinator's metric group when
hive sync is enabled.
+ */
+ private FlinkHiveSyncMetrics hiveSyncMetrics;
+
/**
* The table state.
*/
@@ -457,6 +464,8 @@ public class StreamWriteOperatorCoordinator
.threadFactory(getThreadFactory("hive-sync"))
.waitForTasksFinish(true).build();
this.hiveSyncContext = HiveSyncContext.create(conf, this.storageConf);
+ this.hiveSyncMetrics = new FlinkHiveSyncMetrics(context.metricGroup());
+ this.hiveSyncMetrics.registerMetrics();
}
private void syncHiveAsync() {
@@ -476,9 +485,18 @@ public class StreamWriteOperatorCoordinator
* Sync hoodie table metadata to Hive metastore.
*/
public void doSyncHive() {
+ HoodieTimer timer = HoodieTimer.start();
try (HiveSyncTool syncTool = hiveSyncContext.hiveSyncTool()) {
+ if (!syncTool.isSyncClientInitialized()) {
+ hiveSyncMetrics.markSyncFailed(timer.endTimer());
+ return;
+ }
syncTool.syncHoodieTable();
+ } catch (Throwable t) {
+ hiveSyncMetrics.markSyncFailed(timer.endTimer());
+ throw t;
}
+ hiveSyncMetrics.markSyncSucceeded(timer.endTimer());
}
private void scheduleTableServices(Boolean committed) {
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/metrics/TestFlinkHiveSyncMetrics.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/metrics/TestFlinkHiveSyncMetrics.java
new file mode 100644
index 000000000000..536999adb79e
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/metrics/TestFlinkHiveSyncMetrics.java
@@ -0,0 +1,201 @@
+/*
+ * 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.metrics;
+
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+import static
org.apache.hudi.metrics.FlinkHiveSyncMetrics.HIVE_SYNC_DURATION_MS;
+import static
org.apache.hudi.metrics.FlinkHiveSyncMetrics.HIVE_SYNC_FAILURE_COUNT;
+import static
org.apache.hudi.metrics.FlinkHiveSyncMetrics.HIVE_SYNC_LAST_SUCCESS_TIME_MS;
+import static
org.apache.hudi.metrics.FlinkHiveSyncMetrics.HIVE_SYNC_SUCCESS_COUNT;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for {@link FlinkHiveSyncMetrics}.
+ */
+class TestFlinkHiveSyncMetrics {
+
+ private CapturingMetricGroup metricGroup;
+ private FlinkHiveSyncMetrics metrics;
+
+ @BeforeEach
+ void setUp() {
+ metricGroup = new CapturingMetricGroup();
+ metrics = new FlinkHiveSyncMetrics(metricGroup);
+ metrics.registerMetrics();
+ }
+
+ @Test
+ void testRegisterMetrics() {
+ assertNotNull(metricGroup.counters.get(HIVE_SYNC_SUCCESS_COUNT));
+ assertNotNull(metricGroup.counters.get(HIVE_SYNC_FAILURE_COUNT));
+ assertNotNull(metricGroup.histograms.get(HIVE_SYNC_DURATION_MS));
+ assertNotNull(metricGroup.gauges.get(HIVE_SYNC_LAST_SUCCESS_TIME_MS));
+ }
+
+ @Test
+ void testInitialValues() {
+ assertEquals(0, successCount());
+ assertEquals(0, failureCount());
+ assertEquals(0, duration().getCount());
+ assertEquals(0L, lastSuccessTime());
+ }
+
+ @Test
+ void testMarkSyncSucceeded() {
+ long before = System.currentTimeMillis();
+ metrics.markSyncSucceeded(42L);
+ long after = System.currentTimeMillis();
+
+ assertEquals(1, successCount());
+ assertEquals(0, failureCount());
+ assertEquals(1, duration().getCount());
+ assertEquals(42L, duration().getStatistics().getMax());
+ long lastSuccessTime = lastSuccessTime();
+ assertTrue(lastSuccessTime >= before && lastSuccessTime <= after,
+ "Last success time " + lastSuccessTime + " should be within [" +
before + ", " + after + "]");
+ }
+
+ @Test
+ void testMarkSyncFailed() {
+ metrics.markSyncFailed(1000L);
+
+ assertEquals(0, successCount());
+ assertEquals(1, failureCount());
+ assertEquals(1, duration().getCount());
+ assertEquals(1000L, duration().getStatistics().getMax());
+ assertEquals(0L, lastSuccessTime(), "A failed sync must not set the last
success time");
+ }
+
+ @Test
+ void testFailureKeepsLastSuccessTime() {
+ metrics.markSyncSucceeded(10L);
+ long lastSuccessTime = lastSuccessTime();
+
+ metrics.markSyncFailed(20L);
+
+ assertEquals(1, successCount());
+ assertEquals(1, failureCount());
+ assertEquals(2, duration().getCount());
+ assertEquals(10L, duration().getStatistics().getMin());
+ assertEquals(20L, duration().getStatistics().getMax());
+ assertEquals(lastSuccessTime, lastSuccessTime());
+ }
+
+ @Test
+ void testDurationWindowKeepsLatestSamples() {
+ for (int i = 1; i <= 150; i++) {
+ metrics.markSyncSucceeded(i);
+ }
+
+ assertEquals(150, successCount());
+ assertEquals(150, duration().getCount(), "The count covers every sync, not
just the window");
+ assertEquals(100, duration().getStatistics().size());
+ assertEquals(51L, duration().getStatistics().getMin());
+ assertEquals(150L, duration().getStatistics().getMax());
+ }
+
+ @Test
+ void testConcurrentUpdates() throws Exception {
+ int threads = 2;
+ int syncsPerThread = 10_000;
+ ExecutorService executor = Executors.newFixedThreadPool(threads);
+ try {
+ CountDownLatch start = new CountDownLatch(1);
+ Future<?> succeeding = executor.submit(() -> {
+ start.await();
+ for (int i = 0; i < syncsPerThread; i++) {
+ metrics.markSyncSucceeded(1L);
+ }
+ return null;
+ });
+ Future<?> failing = executor.submit(() -> {
+ start.await();
+ for (int i = 0; i < syncsPerThread; i++) {
+ metrics.markSyncFailed(1L);
+ }
+ return null;
+ });
+ start.countDown();
+ succeeding.get(30, TimeUnit.SECONDS);
+ failing.get(30, TimeUnit.SECONDS);
+ } finally {
+ executor.shutdownNow();
+ }
+
+ assertEquals(syncsPerThread, successCount());
+ assertEquals(syncsPerThread, failureCount());
+ assertEquals(2L * syncsPerThread, duration().getCount());
+ }
+
+ private long successCount() {
+ return metricGroup.counters.get(HIVE_SYNC_SUCCESS_COUNT).getCount();
+ }
+
+ private long failureCount() {
+ return metricGroup.counters.get(HIVE_SYNC_FAILURE_COUNT).getCount();
+ }
+
+ private Histogram duration() {
+ return metricGroup.histograms.get(HIVE_SYNC_DURATION_MS);
+ }
+
+ private long lastSuccessTime() {
+ return (Long)
metricGroup.gauges.get(HIVE_SYNC_LAST_SUCCESS_TIME_MS).getValue();
+ }
+
+ private static class CapturingMetricGroup extends UnregisteredMetricsGroup {
+ private final Map<String, Counter> counters = new HashMap<>();
+ private final Map<String, Histogram> histograms = new HashMap<>();
+ private final Map<String, Gauge<?>> gauges = new HashMap<>();
+
+ @Override
+ public <C extends Counter> C counter(String name, C counter) {
+ counters.put(name, counter);
+ return counter;
+ }
+
+ @Override
+ public <H extends Histogram> H histogram(String name, H histogram) {
+ histograms.put(name, histogram);
+ return histogram;
+ }
+
+ @Override
+ public <T, G extends Gauge<T>> G gauge(String name, G gauge) {
+ gauges.put(name, gauge);
+ return gauge;
+ }
+ }
+}
diff --git
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java
index 7dd73af9dfaa..5a06be17b40a 100644
---
a/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java
+++
b/hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/TestStreamWriteOperatorCoordinator.java
@@ -47,6 +47,7 @@ import org.apache.hudi.configuration.HadoopConfigurations;
import org.apache.hudi.exception.HoodieException;
import org.apache.hudi.exception.MissingSchemaFieldException;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
+import org.apache.hudi.hive.HoodieHiveSyncException;
import org.apache.hudi.metadata.HoodieTableMetadata;
import org.apache.hudi.metadata.MetadataPartitionType;
import org.apache.hudi.sink.event.Correspondent;
@@ -66,6 +67,11 @@ import org.apache.hudi.utils.TestConfigurations;
import org.apache.hudi.utils.TestUtils;
import org.apache.flink.configuration.Configuration;
+import org.apache.flink.metrics.Counter;
+import org.apache.flink.metrics.Gauge;
+import org.apache.flink.metrics.Histogram;
+import org.apache.flink.metrics.groups.OperatorCoordinatorMetricGroup;
+import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
import org.apache.flink.runtime.jobgraph.OperatorID;
import org.apache.flink.runtime.operators.coordination.CoordinationResponse;
import
org.apache.flink.runtime.operators.coordination.MockOperatorCoordinatorContext;
@@ -112,6 +118,7 @@ import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
@@ -604,6 +611,60 @@ public class TestStreamWriteOperatorCoordinator {
assertDoesNotThrow(() -> coordinator.notifyCheckpointComplete(1));
}
+ @Test
+ void testHiveSyncMetricsNotRegisteredWhenHiveSyncDisabled() throws Exception
{
+ coordinator.close();
+ CapturingCoordinatorMetricGroup metricGroup = new
CapturingCoordinatorMetricGroup();
+ coordinator =
startCoordinator(TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()),
metricGroup);
+
+ assertTrue(metricGroup.counters.isEmpty());
+ assertTrue(metricGroup.histograms.isEmpty());
+ assertTrue(metricGroup.gauges.isEmpty());
+ }
+
+ @Test
+ void testHiveSyncFailureUpdatesMetrics() throws Exception {
+ coordinator.close();
+ Configuration conf = getUnreachableMetastoreConf();
+ CapturingCoordinatorMetricGroup metricGroup = new
CapturingCoordinatorMetricGroup();
+ coordinator = startCoordinator(conf, metricGroup);
+
+ assertEquals(0L,
metricGroup.gauges.get("hiveSyncLastSuccessTimeMs").getValue());
+
+ assertThrows(HoodieHiveSyncException.class, () ->
coordinator.doSyncHive());
+ assertThrows(HoodieHiveSyncException.class, () ->
coordinator.doSyncHive());
+
+ assertEquals(0,
metricGroup.counters.get("hiveSyncSuccessCount").getCount());
+ assertEquals(2,
metricGroup.counters.get("hiveSyncFailureCount").getCount());
+ assertEquals(2,
metricGroup.histograms.get("hiveSyncDurationMs").getCount());
+ assertEquals(0L,
metricGroup.gauges.get("hiveSyncLastSuccessTimeMs").getValue());
+ }
+
+ @Test
+ void testHiveSyncWithIgnoredClientFailureCountsAsFailure() throws Exception {
+ coordinator.close();
+ Configuration conf = getUnreachableMetastoreConf();
+ conf.set(FlinkOptions.HIVE_SYNC_IGNORE_EXCEPTIONS, true);
+ CapturingCoordinatorMetricGroup metricGroup = new
CapturingCoordinatorMetricGroup();
+ coordinator = startCoordinator(conf, metricGroup);
+
+ assertDoesNotThrow(() -> coordinator.doSyncHive());
+
+ assertEquals(0,
metricGroup.counters.get("hiveSyncSuccessCount").getCount());
+ assertEquals(1,
metricGroup.counters.get("hiveSyncFailureCount").getCount());
+ assertEquals(1,
metricGroup.histograms.get("hiveSyncDurationMs").getCount());
+ assertEquals(0L,
metricGroup.gauges.get("hiveSyncLastSuccessTimeMs").getValue());
+ }
+
+ private Configuration getUnreachableMetastoreConf() {
+ Configuration conf =
TestConfigurations.getDefaultConf(tempFile.getAbsolutePath());
+ conf.set(FlinkOptions.HIVE_SYNC_ENABLED, true);
+ // nothing listens on port 1, so creating the metastore client fails on
its only attempt
+ conf.set(FlinkOptions.HIVE_SYNC_METASTORE_URIS, "thrift://localhost:1");
+ conf.setString(HadoopConfigurations.HADOOP_PREFIX +
"hive.metastore.connect.retries", "0");
+ return conf;
+ }
+
@Test
void testSyncMetadataTable() throws Exception {
// reset
@@ -998,6 +1059,20 @@ public class TestStreamWriteOperatorCoordinator {
return coordinator;
}
+ private static StreamWriteOperatorCoordinator startCoordinator(
+ Configuration conf, OperatorCoordinatorMetricGroup metricGroup) throws
Exception {
+ MockOperatorCoordinatorContext coordinatorContext = new
MockOperatorCoordinatorContext(new OperatorID(), 1) {
+ @Override
+ public OperatorCoordinatorMetricGroup metricGroup() {
+ return metricGroup;
+ }
+ };
+ StreamWriteOperatorCoordinator coordinator = new
StreamWriteOperatorCoordinator(conf, coordinatorContext);
+ coordinator.start();
+ setSynchronousExecutors(coordinator);
+ return coordinator;
+ }
+
private static void setSynchronousExecutors(StreamWriteOperatorCoordinator
coordinator) throws Exception {
MockOperatorCoordinatorContext coordinatorContext =
(MockOperatorCoordinatorContext) coordinator.getContext();
coordinator.setExecutor(new MockCoordinatorExecutor(coordinatorContext));
@@ -1009,6 +1084,34 @@ public class TestStreamWriteOperatorCoordinator {
return new StreamWriteOperatorCoordinator(conf, coordinatorContext);
}
+ /**
+ * A coordinator metric group that keeps what is registered on it.
+ */
+ private static final class CapturingCoordinatorMetricGroup extends
UnregisteredMetricsGroup
+ implements OperatorCoordinatorMetricGroup {
+ private final Map<String, Counter> counters = new HashMap<>();
+ private final Map<String, Histogram> histograms = new HashMap<>();
+ private final Map<String, Gauge<?>> gauges = new HashMap<>();
+
+ @Override
+ public <C extends Counter> C counter(String name, C counter) {
+ counters.put(name, counter);
+ return counter;
+ }
+
+ @Override
+ public <H extends Histogram> H histogram(String name, H histogram) {
+ histograms.put(name, histogram);
+ return histogram;
+ }
+
+ @Override
+ public <T, G extends Gauge<T>> G gauge(String name, G gauge) {
+ gauges.put(name, gauge);
+ return gauge;
+ }
+ }
+
/**
* A real single-thread instant-request worker whose submitted creation task
blocks on a latch before
* running, to deterministically delay instant creation (simulating a held
table lock).
diff --git
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java
index 4c8aca377089..f0e3ee7da2a9 100644
---
a/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java
+++
b/hudi-sync/hudi-hive-sync/src/main/java/org/apache/hudi/hive/HiveSyncTool.java
@@ -190,6 +190,14 @@ public class HiveSyncTool extends HoodieSyncTool
implements AutoCloseable {
}
}
+ /**
+ * Returns whether the metastore client was created. It is not when creating
it failed and
+ * {@code hoodie.datasource.hive_sync.ignore_exceptions} is set, in which
case a sync does nothing.
+ */
+ public boolean isSyncClientInitialized() {
+ return syncClient != null;
+ }
+
protected void doSync() {
// create database if needed
checkAndCreateDatabase();
