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 108d72fe15bb9b22dfd0ce2366e6740bc3c1b833 Author: Danny Chan <[email protected]> AuthorDate: Tue Sep 29 17:26:42 2026 +0800 perf(flink): refresh coordinator meta client in place (#20134) Backport only the coordinator implementation and corresponding test changes. Refresh the timeline and table properties using the existing meta client, and remove the unsupported secondary-index validation test. (cherry picked from commit 5550b458563131bd0bfb331ee364944394fa5ddd) --- .../hudi/sink/StreamWriteOperatorCoordinator.java | 4 ++-- .../sink/TestStreamWriteOperatorCoordinator.java | 25 ---------------------- 2 files changed, 2 insertions(+), 27 deletions(-) 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 07cbe1e1db3e..983e8d3314a2 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,8 +490,8 @@ public class StreamWriteOperatorCoordinator } private String startInstant() { - // Refresh table properties and index definitions as well as the timeline before validating the new write. - this.metaClient = HoodieTableMetaClient.reload(this.metaClient); + // Refresh table properties and the timeline before validating the new write. + this.metaClient.reload(); // Validate the write and refresh the last txn metadata. this.writeClient.preTxn(tableState.operationType, this.metaClient); // put the assignment in front of metadata generation, 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 b8691bdc9c30..17712c852d5e 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 @@ -24,7 +24,6 @@ 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; @@ -43,9 +42,7 @@ 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; @@ -201,28 +198,6 @@ public class TestStreamWriteOperatorCoordinator { 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))
