You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by da...@apache.org on 2026/10/08 12:07:46 UTC
(hudi) 16/17: perf(flink): validate schema and table properties once per instant (#20122)
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 <yu...@gmail.com>
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());