fix(flink): Fix reflection ctor signature for AwsGlueCatalogSyncTool in HiveSyncContext

hudi-aws/src/test/java/org/apache/hudi/aws/sync/TestAwsGlueSyncTool.java

@@ -21,9 +21,11 @@
 import org.apache.hudi.aws.testutils.GlueTestUtil;
 import org.apache.hudi.common.config.TypedProperties;
 import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ReflectionUtils;
 import org.apache.hudi.sync.common.HoodieSyncTool;
 import org.apache.hudi.sync.common.util.SyncUtilHelpers;
 
+import org.apache.hadoop.conf.Configuration;
 import org.junit.jupiter.api.AfterAll;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
@@ -38,6 +40,7 @@
 import software.amazon.awssdk.services.sts.model.GetCallerIdentityResponse;
 
 import java.io.IOException;
+import java.util.Properties;
 
 import static org.apache.hudi.aws.testutils.GlueTestUtil.getHadoopConf;
 import static org.apache.hudi.aws.testutils.GlueTestUtil.glueSyncProps;
@@ -105,4 +108,32 @@ void validateInitThroughSyncTool() throws Exception {
       syncTool.close();
     }
   }
+
+  /**
+   * Verifies the 3-arg {@code (Properties, Configuration, Option)} reflection path used by Flink's
+   * {@code HiveSyncContext#hiveSyncTool()} can fully instantiate {@link AwsGlueCatalogSyncTool}.
+   */
+  @Test
+  void validateInitThroughHiveSyncContextReflectionSignature() throws Exception {
+    try (MockedStatic<GlueAsyncClient> mockedStatic = mockStatic(GlueAsyncClient.class);
+         MockedStatic<StsClient> mockedStsStatic = mockStatic(StsClient.class)) {
+      GlueAsyncClientBuilder builder = mock(GlueAsyncClientBuilder.class);
+      mockedStatic.when(GlueAsyncClient::builder).thenReturn(builder);
+      when(builder.credentialsProvider(any())).thenReturn(builder);
+      GlueAsyncClient mockClient = mock(GlueAsyncClient.class);
+      when(builder.build()).thenReturn(mockClient);
+      StsClient mockSts = mock(StsClient.class);
+      mockedStsStatic.when(StsClient::create).thenReturn(mockSts);
+      when(mockSts.getCallerIdentity(GetCallerIdentityRequest.builder().build()))
+          .thenReturn(GetCallerIdentityResponse.builder().account("").build());
+
+      Object syncTool = ReflectionUtils.loadClass(
+          AwsGlueCatalogSyncTool.class.getName(),
+          new Class<?>[] {Properties.class, Configuration.class, Option.class},
+          glueSyncProps, getHadoopConf(), Option.empty());
+      assertTrue(syncTool instanceof AwsGlueCatalogSyncTool,
+          "Reflection through the HiveSyncContext signature must yield an AwsGlueCatalogSyncTool instance");
+      ((AwsGlueCatalogSyncTool) syncTool).close();
+    }
+  }
 }

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/utils/HiveSyncContext.java

@@ -19,6 +19,8 @@
 package org.apache.hudi.sink.utils;
 
 import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.table.HoodieTableMetaClient;
+import org.apache.hudi.common.util.Option;
 import org.apache.hudi.common.util.ReflectionUtils;
 import org.apache.hudi.configuration.FlinkOptions;
 import org.apache.hudi.configuration.HadoopConfigurations;
@@ -78,8 +80,8 @@ public HiveSyncTool hiveSyncTool() {
     HiveSyncMode syncMode = HiveSyncMode.of(props.getProperty(HIVE_SYNC_MODE.key()));
     if (syncMode == HiveSyncMode.GLUE) {
       return ((HiveSyncTool) ReflectionUtils.loadClass(AWS_GLUE_CATALOG_SYNC_TOOL_CLASS,
-          new Class<?>[] {Properties.class, org.apache.hadoop.conf.Configuration.class},
-          props, hiveConf));
+          new Class<?>[] {Properties.class, org.apache.hadoop.conf.Configuration.class, Option.class},
+          props, hiveConf, Option.<HoodieTableMetaClient>empty()));
     }
     return new HiveSyncTool(props, hiveConf);
   }

hudi-flink-datasource/hudi-flink/src/test/java/org/apache/hudi/sink/utils/TestHiveSyncContext.java

@@ -18,6 +18,8 @@
 
 package org.apache.hudi.sink.utils;
 
+import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.ReflectionUtils;
 import org.apache.hudi.configuration.FlinkOptions;
 import org.apache.hudi.hive.HiveSyncConfig;
 
@@ -66,4 +68,18 @@ void testOptionWithoutShortcutKey() {
     Properties props3 = HiveSyncContext.buildSyncConfig(configuration3);
     assertTrue(Boolean.parseBoolean(props3.getProperty(HiveSyncConfig.HIVE_CREATE_MANAGED_TABLE.key(), "false")));
   }
+
+  /**
+   * Pins the constructor signature {@link HiveSyncContext#hiveSyncTool()} relies on via reflection.
+   * End-to-end instantiation is covered in {@code hudi-aws}'s {@code TestAwsGlueSyncTool}.
+   */
+  @Test
+  void testAwsGlueSyncToolReflectionConstructorExists() {
+    assertTrue(
+        ReflectionUtils.hasConstructor(
+            HiveSyncContext.AWS_GLUE_CATALOG_SYNC_TOOL_CLASS,
+            new Class<?>[] {Properties.class, org.apache.hadoop.conf.Configuration.class, Option.class}),
+        "AwsGlueCatalogSyncTool must expose the constructor used by HiveSyncContext#hiveSyncTool() "
+            + "via reflection; otherwise Flink GLUE sync fails with NoSuchMethodException.");
+  }
 }