fix rescale configs

psc-flink/src/main/java/com/pinterest/flink/streaming/connectors/psc/table/PscTableCommonUtils.java

@@ -20,6 +20,7 @@
 
 import com.pinterest.psc.common.TopicUri;
 import com.pinterest.psc.config.PscConfiguration;
+import com.pinterest.psc.config.PscConfigurationUtils;
 import com.pinterest.psc.metadata.TopicUriMetadata;
 import com.pinterest.psc.metadata.client.PscMetadataClient;
 import org.apache.flink.annotation.VisibleForTesting;
@@ -184,12 +185,24 @@ private static int getTopicPartitionCount(List<String> topicUris, Properties psc
 
         PscMetadataClient metadataClient = null;
         try {
-            // Create PSC configuration from properties
-            PscConfiguration pscConfig = new PscConfiguration();
-            for (String key : pscProperties.stringPropertyNames()) {
-                pscConfig.setProperty(key, pscProperties.getProperty(key));
+            // Create a copy of properties to avoid modifying the original
+            Properties metadataClientProps = new Properties();
+            metadataClientProps.putAll(pscProperties);
+
+            // Set the required psc.metadata.client.id if not already present.
+            // This is required by PscMetadataClient validation (see PscConfigurationInternal.validateMetadataClientConfiguration).
+            // PscSourceEnumerator also sets this before creating its metadata client.
+            if (!metadataClientProps.containsKey(PscConfiguration.PSC_METADATA_CLIENT_ID)) {
+                metadataClientProps.setProperty(
+                        PscConfiguration.PSC_METADATA_CLIENT_ID, 
+                        "psc-table-partition-count-query");
             }
 
+            // Convert properties to PSC configuration.
+            // When passed to PscMetadataClient, it will be wrapped in PscConfigurationInternal
+            // which loads psc.conf defaults and validates the configuration.
+            PscConfiguration pscConfig = PscConfigurationUtils.propertiesToPscConfiguration(metadataClientProps);
+
             metadataClient = new PscMetadataClient(pscConfig);
 
             int minPartitionCount = Integer.MAX_VALUE;

psc-flink/src/test/java/com/pinterest/flink/streaming/connectors/psc/table/PscTableCommonUtilsTest.java

@@ -18,6 +18,10 @@
 
 package com.pinterest.flink.streaming.connectors.psc.table;
 
+import com.pinterest.psc.config.PscConfiguration;
+import com.pinterest.psc.config.PscConfigurationInternal;
+import com.pinterest.psc.config.PscConfigurationUtils;
+import com.pinterest.psc.exception.startup.ConfigurationException;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.table.api.config.ExecutionConfigOptions;
 import org.junit.After;
@@ -30,6 +34,7 @@
 
 import static com.pinterest.flink.streaming.connectors.psc.table.PscConnectorOptions.SCAN_ENABLE_RESCALE;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
 
 /**
  * Comprehensive test suite for PscTableCommonUtils, focusing on rescale decision logic
@@ -355,5 +360,29 @@ public void testProviderResetRestoresDefaultBehavior() {
             tableOptions, globalConfig, topicUris, pscProperties, null);
         assertThat(resultAfterReset).isFalse(); // partition count = -1, fail-safe
     }
+
+    // ============================================
+    // Tests validating PscMetadataClient configuration fix
+    // ============================================
+
+    @Test
+    public void testMetadataClientValidationPassesWithClientId() {
+        // This test proves the FIX works:
+        // With psc.metadata.client.id set, PscConfigurationInternal validation passes
+
+        // Given: Properties WITH psc.metadata.client.id (matching our fix)
+        Properties propsWithClientId = new Properties();
+        propsWithClientId.setProperty(
+            PscConfiguration.PSC_METADATA_CLIENT_ID, 
+            "psc-table-partition-count-query"
+        );
+
+        PscConfiguration pscConfig = PscConfigurationUtils.propertiesToPscConfiguration(propsWithClientId);
+
+        // When/Then: Creating PscConfigurationInternal for metadata client does NOT throw
+        assertThatCode(() -> 
+            new PscConfigurationInternal(pscConfig, PscConfigurationInternal.PSC_CLIENT_TYPE_METADATA)
+        ).doesNotThrowAnyException();
+    }
 }