This is an automated email from the ASF dual-hosted git repository. danny0405 pushed a commit to branch release-1.1.2 in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 98f75e0be4e3c7a345096c5409b898245839dbf3 Author: Danny Chan <[email protected]> AuthorDate: Mon Sep 28 18:34:26 2026 +0800 perf(flink): validate schema and table properties once per instant (#20122) * perf(flink): validate schema and table properties once per instant (cherry picked from commit c74939e3b1ed026d93426932e82192d9291c4570) --- .../apache/hudi/client/BaseHoodieWriteClient.java | 22 ++-- .../apache/hudi/client/HoodieFlinkWriteClient.java | 18 ++-- .../apache/hudi/client/TestFlinkWriteClient.java | 75 ++++++++++++++ .../hudi/sink/StreamWriteOperatorCoordinator.java | 6 +- .../sink/TestStreamWriteOperatorCoordinator.java | 113 +++++++++++++++++++++ 5 files changed, 211 insertions(+), 23 deletions(-) diff --git a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java index 183b57eb5ee0..8804ddd43c3a 100644 --- a/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java +++ b/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/client/BaseHoodieWriteClient.java @@ -1348,10 +1348,9 @@ public abstract class BaseHoodieWriteClient<T, I, K, O> extends BaseHoodieClient } /** - * Performs necessary bootstrapping operations (for ex, validating whether Metadata Table has to be bootstrapped). + * Performs necessary bootstrapping operations and validates the table properties. * - * <p>NOTE: THIS OPERATION IS EXECUTED UNDER LOCK, THEREFORE SHOULD AVOID ANY OPERATIONS - * NOT REQUIRING EXTERNAL SYNCHRONIZATION + * <p>Upgrade and metadata table initialization execute under lock. Table properties are validated afterward. * * @param metaClient instance of {@link HoodieTableMetaClient} * @param instantTime current inflight instant time @@ -1363,14 +1362,14 @@ public abstract class BaseHoodieWriteClient<T, I, K, O> extends BaseHoodieClient metaClient.getTableType()), instantTime.get())); } boolean requiresInitTable = needsUpgrade(metaClient) || config.isMetadataTableEnabled(); - if (!requiresInitTable) { - return; + if (requiresInitTable) { + executeUsingTxnManager(ownerInstant, () -> { + tryUpgrade(metaClient, instantTime); + // TODO: this also does MT table management.. + initMetadataTable(instantTime, metaClient); + }); } - executeUsingTxnManager(ownerInstant, () -> { - tryUpgrade(metaClient, instantTime); - // TODO: this also does MT table management.. - initMetadataTable(instantTime, metaClient); - }); + validateAgainstTableProperties(metaClient.getTableConfig(), config); } /** @@ -1410,9 +1409,6 @@ public abstract class BaseHoodieWriteClient<T, I, K, O> extends BaseHoodieClient doInitTable(operationType, metaClient, instantTime); HoodieTable table = createTable(config, metaClient); - // Validate table properties - validateAgainstTableProperties(table.getMetaClient().getTableConfig(), config); - switch (operationType) { case INSERT: case INSERT_PREPPED: diff --git a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java index 5c40e4def972..d05b7a0f7ee2 100644 --- a/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java +++ b/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/client/HoodieFlinkWriteClient.java @@ -69,6 +69,9 @@ import java.util.stream.Collectors; * <p>The client is used both on driver (for starting/committing transactions) * and executor (for writing dataset). * + * <p>Table property and schema validation for streaming writes is performed by the operator coordinator before + * making each new instant available to the write tasks. + * * @param <T> type of the payload */ @SuppressWarnings("checkstyle:LineLength") @@ -158,7 +161,6 @@ public class HoodieFlinkWriteClient<T> public List<WriteStatus> upsert(Iterator<HoodieRecord<T>> records, BucketInfo bucketInfo, String instantTime) { HoodieTable<T, List<HoodieRecord<T>>, List<HoodieKey>, List<WriteStatus>> table = initTable(WriteOperationType.UPSERT, Option.ofNullable(instantTime)); - table.validateUpsertSchema(); preWrite(instantTime, WriteOperationType.UPSERT, table.getMetaClient()); HoodieWriteMetadata<List<WriteStatus>> result; @@ -176,7 +178,6 @@ public class HoodieFlinkWriteClient<T> // only used for metadata table, the upsert happens in single thread HoodieTable<T, List<HoodieRecord<T>>, List<HoodieKey>, List<WriteStatus>> table = initTable(WriteOperationType.UPSERT, Option.ofNullable(instantTime)); - table.validateUpsertSchema(); preWrite(instantTime, WriteOperationType.UPSERT_PREPPED, table.getMetaClient()); Map<String, List<HoodieRecord<T>>> preppedRecordsByFileId = preppedRecords.stream().parallel() .collect(Collectors.groupingBy(r -> r.getCurrentLocation().getFileId())); @@ -200,7 +201,6 @@ public class HoodieFlinkWriteClient<T> public List<WriteStatus> insert(Iterator<HoodieRecord<T>> records, BucketInfo bucketInfo, String instantTime) { HoodieTable<T, List<HoodieRecord<T>>, List<HoodieKey>, List<WriteStatus>> table = initTable(WriteOperationType.INSERT, Option.ofNullable(instantTime)); - table.validateInsertSchema(); preWrite(instantTime, WriteOperationType.INSERT, table.getMetaClient()); HoodieWriteMetadata<List<WriteStatus>> result; @@ -217,7 +217,6 @@ public class HoodieFlinkWriteClient<T> public List<WriteStatus> insertOverwrite(Iterator<HoodieRecord<T>> records, BucketInfo bucketInfo, String instantTime) { HoodieTable<T, List<HoodieRecord<T>>, List<HoodieKey>, List<WriteStatus>> table = initTable(WriteOperationType.INSERT_OVERWRITE, Option.ofNullable(instantTime)); - table.validateInsertSchema(); preWrite(instantTime, WriteOperationType.INSERT_OVERWRITE, table.getMetaClient()); // create the write handle if not exists HoodieWriteMetadata<List<WriteStatus>> result; @@ -230,7 +229,6 @@ public class HoodieFlinkWriteClient<T> @Override public List<WriteStatus> insertOverwriteTable(Iterator<HoodieRecord<T>> records, BucketInfo bucketInfo, String instantTime) { HoodieTable table = initTable(WriteOperationType.INSERT_OVERWRITE_TABLE, Option.ofNullable(instantTime)); - table.validateInsertSchema(); preWrite(instantTime, WriteOperationType.INSERT_OVERWRITE_TABLE, table.getMetaClient()); // create the write handle if not exists HoodieWriteMetadata<List<WriteStatus>> result; @@ -260,7 +258,6 @@ public class HoodieFlinkWriteClient<T> // only used for metadata table, the bulk_insert happens in single JVM process HoodieTable<T, List<HoodieRecord<T>>, List<HoodieKey>, List<WriteStatus>> table = initTable(WriteOperationType.BULK_INSERT_PREPPED, Option.ofNullable(instantTime)); - table.validateInsertSchema(); preWrite(instantTime, WriteOperationType.BULK_INSERT_PREPPED, table.getMetaClient()); Map<String, List<HoodieRecord<T>>> preppedRecordsByFileId = preppedRecords.stream().parallel() .collect(Collectors.groupingBy(r -> r.getCurrentLocation().getFileId())); @@ -319,10 +316,15 @@ public class HoodieFlinkWriteClient<T> } /** - * Refresh the last transaction metadata, + * Validate the table properties and write schema, and refresh the last transaction metadata, * should be called before the Driver starts a new transaction with a reloaded metaclient. */ public void preTxn(WriteOperationType operationType, HoodieTableMetaClient metaClient) { + // Validate once on the coordinator using its refreshed timeline, before writers receive the instant. + validateAgainstTableProperties(metaClient.getTableConfig(), config); + if (!metaClient.isMetadataTable() && (WriteOperationType.isChangingRecords(operationType) || WriteOperationType.isInsert(operationType))) { + createTable(config, metaClient).validateSchema(); + } if (txnManager.isLockRequired() && config.needResolveWriteConflict(operationType, metaClient.isMetadataTable(), config, metaClient.getTableConfig())) { this.lastCompletedTxnAndMetadata = TransactionUtils.getLastCompletedTxnInstantAndMetadata(metaClient); this.pendingInflightAndRequestedInstants = TransactionUtils.getInflightAndRequestedInstants(metaClient); @@ -398,6 +400,8 @@ public class HoodieFlinkWriteClient<T> // no need to execute the upgrade/downgrade on each write in streaming. // flink performs metadata table bootstrap on the coordinator when it starts up. + + // flink validates table properties on the coordinator in preTxn. } /** diff --git a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/TestFlinkWriteClient.java b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/TestFlinkWriteClient.java index 0b532bbe5570..1bc2efafea57 100644 --- a/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/TestFlinkWriteClient.java +++ b/hudi-client/hudi-flink-client/src/test/java/org/apache/hudi/client/TestFlinkWriteClient.java @@ -22,6 +22,10 @@ package org.apache.hudi.client; import org.apache.hudi.client.heartbeat.HoodieHeartbeatClient; import org.apache.hudi.common.engine.EngineType; import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy; +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.util.Option; import org.apache.hudi.config.HoodieCleanConfig; import org.apache.hudi.config.HoodieWriteConfig; import org.apache.hudi.exception.HoodieException; @@ -32,15 +36,25 @@ import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.EnumSource; import org.junit.jupiter.params.provider.ValueSource; import java.io.IOException; +import java.util.Collections; +import java.util.Properties; import java.util.concurrent.atomic.AtomicBoolean; 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.clearInvocations; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; public class TestFlinkWriteClient extends HoodieFlinkClientTestHarness { @@ -56,6 +70,67 @@ public class TestFlinkWriteClient extends HoodieFlinkClientTestHarness { cleanupResources(); } + @Test + void testTablePropertyValidationRunsOnlyInPreTxn() throws IOException { + Properties updatedProperties = new Properties(); + updatedProperties.setProperty(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false"); + HoodieTableConfig.update(metaClient.getStorage(), metaClient.getMetaPath(), updatedProperties); + metaClient = HoodieTableMetaClient.reload(metaClient); + HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder() + .withPath(metaClient.getBasePath()) + .withEngineType(EngineType.FLINK) + .withEmbeddedTimelineServerEnabled(false) + .build(); + writeClient = new HoodieFlinkWriteClient(context, writeConfig); + + // Per-bucket table initialization defers validation to the coordinator. + writeClient.initTable(WriteOperationType.UPSERT, Option.empty()); + HoodieException exception = assertThrows(HoodieException.class, + () -> writeClient.preTxn(WriteOperationType.UPSERT, metaClient)); + assertTrue(exception.getMessage().contains(HoodieTableConfig.POPULATE_META_FIELDS.key())); + } + + @Test + void testPreTxnSkipsSchemaValidationForMetadataTableDeletePrepped() { + HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder() + .withPath(metaClient.getBasePath()) + .withEngineType(EngineType.FLINK) + .withEmbeddedTimelineServerEnabled(false) + .build(); + writeClient = spy(new HoodieFlinkWriteClient(context, writeConfig)); + HoodieTableMetaClient metadataMetaClient = spy(metaClient); + doReturn(true).when(metadataMetaClient).isMetadataTable(); + writeClient.preTxn(WriteOperationType.DELETE_PREPPED, metadataMetaClient); + + verify(writeClient, never()).createTable(writeConfig, metadataMetaClient); + } + + @ParameterizedTest + @EnumSource(value = WriteOperationType.class, names = {"UPSERT_PREPPED", "BULK_INSERT_PREPPED"}) + void testPreppedSchemaValidationRunsOnlyInPreTxn(WriteOperationType operationType) { + HoodieWriteConfig writeConfig = HoodieWriteConfig.newBuilder() + .withPath(metaClient.getBasePath()) + .withEngineType(EngineType.FLINK) + .withEmbeddedTimelineServerEnabled(false) + .build(); + writeClient = spy(new HoodieFlinkWriteClient(context, writeConfig)); + HoodieTable table = spy(writeClient.getHoodieTable()); + doReturn(table).when(writeClient).createTable(eq(writeConfig), any(HoodieTableMetaClient.class)); + + writeClient.preTxn(operationType, metaClient); + verify(writeClient).validateAgainstTableProperties(metaClient.getTableConfig(), writeConfig); + verify(table).validateSchema(); + + clearInvocations(table, writeClient); + if (operationType == WriteOperationType.UPSERT_PREPPED) { + writeClient.upsertPreppedRecords(Collections.emptyList(), "001"); + } else { + writeClient.bulkInsertPreppedRecords(Collections.emptyList(), "001", Option.empty()); + } + verify(table, never()).validateSchema(); + verify(writeClient, never()).validateAgainstTableProperties(any(), any()); + } + @ParameterizedTest @ValueSource(booleans = {true, false}) public void testWriteClientAndTableServiceClientWithTimelineServer( 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 22ef43e40a35..07cbe1e1db3e 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 @@ -490,9 +490,9 @@ public class StreamWriteOperatorCoordinator } private String startInstant() { - // refresh the meta client which is reused - metaClient.reloadActiveTimeline(); - // refresh the last txn metadata + // Refresh table properties and index definitions as well as the timeline before validating the new write. + this.metaClient = HoodieTableMetaClient.reload(this.metaClient); + // Validate the write and refresh the last txn metadata. this.writeClient.preTxn(tableState.operationType, this.metaClient); // put the assignment in front of metadata generation, // because the instant request from write task is asynchronous. 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 9815878eff99..b8691bdc9c30 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 @@ -18,24 +18,34 @@ package org.apache.hudi.sink; +import org.apache.hudi.client.HoodieFlinkWriteClient; import org.apache.hudi.client.WriteStatus; import org.apache.hudi.client.heartbeat.HoodieHeartbeatClient; import org.apache.hudi.common.fs.FSUtils; +import org.apache.hudi.common.model.HoodieCommitMetadata; import org.apache.hudi.common.model.HoodieFailedWritesCleaningPolicy; +import org.apache.hudi.common.model.HoodieIndexDefinition; import org.apache.hudi.common.model.HoodieTableType; import org.apache.hudi.common.model.HoodieWriteStat; import org.apache.hudi.common.model.WriteConcurrencyMode; +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.timeline.HoodieActiveTimeline; import org.apache.hudi.common.table.timeline.HoodieInstant; import org.apache.hudi.common.table.timeline.HoodieTimeline; +import org.apache.hudi.common.testutils.HoodieTestTable; import org.apache.hudi.common.testutils.HoodieTestUtils; +import org.apache.hudi.common.util.Option; import org.apache.hudi.config.HoodieCleanConfig; import org.apache.hudi.config.HoodieWriteConfig; import org.apache.hudi.configuration.FlinkOptions; import org.apache.hudi.configuration.HadoopConfigurations; import org.apache.hudi.exception.HoodieException; +import org.apache.hudi.exception.MissingSchemaFieldException; +import org.apache.hudi.exception.SchemaCompatibilityException; import org.apache.hudi.hadoop.fs.HadoopFSUtils; +import org.apache.hudi.metadata.HoodieIndexVersion; import org.apache.hudi.metadata.HoodieTableMetadata; import org.apache.hudi.sink.event.Correspondent; import org.apache.hudi.sink.event.WriteMetadataEvent; @@ -70,7 +80,9 @@ import org.slf4j.Logger; import java.io.File; import java.io.IOException; +import java.lang.reflect.Field; import java.util.Collections; +import java.util.Properties; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; @@ -127,6 +139,107 @@ public class TestStreamWriteOperatorCoordinator { assertNotEquals(instant, inflight, "Should start a new instant"); } + @ParameterizedTest + @ValueSource(strings = {"upsert", "insert", "bulk_insert", "insert_overwrite", "insert_overwrite_table", "delete"}) + void testValidationOncePerInstant(String operation) throws Exception { + coordinator.close(); + Configuration conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()); + conf.set(FlinkOptions.OPERATION, operation); + coordinator = startCoordinator(conf, 2); + HoodieFlinkWriteClient writeClient = Mockito.spy(coordinator.getWriteClient()); + Field writeClientField = StreamWriteOperatorCoordinator.class.getDeclaredField("writeClient"); + writeClientField.setAccessible(true); + writeClientField.set(coordinator, writeClient); + + String firstInstant = requestInstantTime(1); + assertEquals(firstInstant, requestInstantTime(1)); + Mockito.verify(writeClient, Mockito.times(1)).preTxn(Mockito.eq(WriteOperationType.fromValue(operation)), Mockito.any()); + Mockito.verify(writeClient, Mockito.times(1)).validateAgainstTableProperties(Mockito.any(), Mockito.any()); + + // Complete the first checkpoint so the next request can create a new instant. + coordinator.handleEventFromOperator(0, createOperatorEvent(0, 1, firstInstant, "par1", false, true, 0.1)); + coordinator.handleEventFromOperator(1, createOperatorEvent(1, 1, firstInstant, "par2", false, true, 0.2)); + coordinator.notifyCheckpointComplete(2); + assertNull(coordinator.getEventBuffer(1)); + String nextInstant = requestInstantTime(2); + assertNotEquals(firstInstant, nextInstant); + Mockito.verify(writeClient, Mockito.times(2)).preTxn(Mockito.eq(WriteOperationType.fromValue(operation)), Mockito.any()); + Mockito.verify(writeClient, Mockito.times(2)).validateAgainstTableProperties(Mockito.any(), Mockito.any()); + assertFalse(((MockOperatorCoordinatorContext) coordinator.getContext()).isJobFailed()); + } + + @ParameterizedTest + @ValueSource(strings = {"upsert", "insert", "bulk_insert", "insert_overwrite", "insert_overwrite_table", "delete"}) + void testColumnDropFailsBeforeInstantIsPublished(String operation) throws Exception { + coordinator.close(); + Configuration conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()); + conf.set(FlinkOptions.OPERATION, operation); + coordinator = startCoordinator(conf, 2); + HoodieWriteConfig writeConfig = coordinator.getWriteClient().getConfig(); + assertFalse(writeConfig.shouldValidateAvroSchema()); + assertFalse(writeConfig.shouldAllowAutoEvolutionColumnDrop()); + + // Complete an instant using the original schema, then drop a column for the next instant. + HoodieCommitMetadata metadata = new HoodieCommitMetadata(); + metadata.addMetadata(HoodieCommitMetadata.SCHEMA_KEY, writeConfig.getSchema()); + HoodieTestTable.of(StreamerUtil.createMetaClient(conf)).addCommit("001", Option.of(metadata)); + conf.set(FlinkOptions.SOURCE_AVRO_SCHEMA_PATH, + getClass().getClassLoader().getResource("test_read_schema_dropped_age.avsc").toString()); + writeConfig.setSchema(StreamerUtil.getSourceSchema(conf).toString()); + + assertInstantCreationFails(conf, MissingSchemaFieldException.class, "age"); + } + + @Test + void testTablePropertyChangesAreValidatedBeforeInstantIsPublished() throws Exception { + Configuration conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()); + HoodieTableMetaClient metaClient = StreamerUtil.createMetaClient(conf); + Properties updatedProperties = new Properties(); + updatedProperties.setProperty(HoodieTableConfig.POPULATE_META_FIELDS.key(), "false"); + HoodieTableConfig.update(metaClient.getStorage(), metaClient.getMetaPath(), updatedProperties); + + assertInstantCreationFails(conf, HoodieException.class, HoodieTableConfig.POPULATE_META_FIELDS.key()); + } + + @Test + void testNewSecondaryIndexIsValidatedBeforeInstantIsPublished() throws Exception { + Configuration conf = TestConfigurations.getDefaultConf(tempFile.getAbsolutePath()); + HoodieWriteConfig writeConfig = coordinator.getWriteClient().getConfig(); + HoodieTableMetaClient metaClient = StreamerUtil.createMetaClient(conf); + HoodieCommitMetadata metadata = new HoodieCommitMetadata(); + metadata.addMetadata(HoodieCommitMetadata.SCHEMA_KEY, writeConfig.getSchema()); + HoodieTestTable.of(metaClient).addCommit("001", Option.of(metadata)); + // Simulate an index created after the coordinator has loaded its meta client. + metaClient.buildIndexDefinition(HoodieIndexDefinition.newBuilder() + .withIndexName("secondary_index_age") + .withIndexType("secondary_index") + .withSourceFields(Collections.singletonList("age")) + .withVersion(HoodieIndexVersion.V1) + .build()); + String evolvedSchema = writeConfig.getSchema().replace("\"int\"", "\"long\""); + assertNotEquals(writeConfig.getSchema(), evolvedSchema); + writeConfig.setSchema(evolvedSchema); + + assertInstantCreationFails(conf, SchemaCompatibilityException.class, "secondary_index_age"); + } + + private void assertInstantCreationFails(Configuration conf, Class<? extends Throwable> causeType, String message) throws Exception { + Correspondent.InstantTimeResponse response = CoordinationResponseSerDe.unwrap( + coordinator.handleCoordinationRequest(Correspondent.InstantTimeRequest.getInstance(1)) + .get(1, TimeUnit.SECONDS)); + assertNull(response.getInstant()); + assertNull(coordinator.getEventBuffer(1)); + MockOperatorCoordinatorContext context = (MockOperatorCoordinatorContext) coordinator.getContext(); + assertTrue(context.isJobFailed()); + Throwable cause = context.getJobFailureReason(); + while (cause != null && !(causeType.isInstance(cause) && cause.getMessage() != null && cause.getMessage().contains(message))) { + cause = cause.getCause(); + } + assertNotNull(cause, "Expected " + causeType.getSimpleName() + " containing: " + message); + HoodieTimeline pending = StreamerUtil.createMetaClient(conf).reloadActiveTimeline().filterPendingExcludingCompaction(); + assertTrue(pending.empty(), "Validation must fail before a new instant is created"); + } + @Test public void testTableInitialized() throws IOException { final org.apache.hadoop.conf.Configuration hadoopConf = HadoopConfigurations.getHadoopConf(new Configuration());
