Don't replace environment variables and system properties in get table configs REST API

pinot-common/src/main/java/org/apache/pinot/common/metadata/ZKMetadataProvider.java

@@ -100,8 +100,8 @@ public static boolean createDatabaseConfig(ZkHelixPropertyStore<ZNRecord> proper
   public static boolean setDatabaseConfig(ZkHelixPropertyStore<ZNRecord> propertyStore, DatabaseConfig databaseConfig) {
     String databaseName = databaseConfig.getDatabaseName();
     ZNRecord databaseConfigZNRecord = toZNRecord(databaseConfig);
-    return propertyStore.set(constructPropertyStorePathForDatabaseConfig(databaseName), databaseConfigZNRecord,
-        -1, AccessOption.PERSISTENT);
+    return propertyStore.set(constructPropertyStorePathForDatabaseConfig(databaseName), databaseConfigZNRecord, -1,
+        AccessOption.PERSISTENT);
   }
 
   /**
@@ -441,14 +441,31 @@ public static Map<String, UserConfig> getAllUserInfo(ZkHelixPropertyStore<ZNReco
 
   @Nullable
   public static DatabaseConfig getDatabaseConfig(ZkHelixPropertyStore<ZNRecord> propertyStore, String databaseName) {
-    return toDatabaseConfig(propertyStore.get(constructPropertyStorePathForDatabaseConfig(databaseName), null,
-        AccessOption.PERSISTENT));
+    return toDatabaseConfig(
+        propertyStore.get(constructPropertyStorePathForDatabaseConfig(databaseName), null, AccessOption.PERSISTENT));
   }
 
+  /**
+   * Get the table config for the given table name with type. Any environment variables and system properties will be
+   * replaced with their actual values.
+   */
   @Nullable
   public static TableConfig getTableConfig(ZkHelixPropertyStore<ZNRecord> propertyStore, String tableNameWithType) {
+    return getTableConfig(propertyStore, tableNameWithType, true);
+  }
+
+  /**
+   * Get the table config for the given table name with type
+   *
+   * @param tableNameWithType Table name with type
+   * @param replaceVariables Whether to replace environment variables and system properties with their actual values
+   * @return Table config
+   */
+  @Nullable
+  public static TableConfig getTableConfig(ZkHelixPropertyStore<ZNRecord> propertyStore, String tableNameWithType,
+      boolean replaceVariables) {
     return toTableConfig(propertyStore.get(constructPropertyStorePathForResourceConfig(tableNameWithType), null,
-        AccessOption.PERSISTENT));
+        AccessOption.PERSISTENT), replaceVariables);
   }
 
   /**
@@ -467,14 +484,54 @@ public static ImmutablePair<TableConfig, Integer> getTableConfigWithVersion(
     return ImmutablePair.of(tableConfig, tableConfigStat.getVersion());
   }
 
+  /**
+   * Get the offline table config for the given table name. Any environment variables and system properties will be
+   * replaced with their actual values.
+   *
+   * @param tableName Table name with or without type suffix
+   * @return Table config
+   */
   @Nullable
   public static TableConfig getOfflineTableConfig(ZkHelixPropertyStore<ZNRecord> propertyStore, String tableName) {
-    return getTableConfig(propertyStore, TableNameBuilder.OFFLINE.tableNameWithType(tableName));
+    return getOfflineTableConfig(propertyStore, tableName, true);
   }
 
+  /**
+   * Get the offline table config for the given table name.
+   *
+   * @param tableName Table name with or without type suffix
+   * @param replaceVariables Whether to replace environment variables and system properties with their actual values
+   * @return Table config
+   */
+  @Nullable
+  public static TableConfig getOfflineTableConfig(ZkHelixPropertyStore<ZNRecord> propertyStore, String tableName,
+      boolean replaceVariables) {
+    return getTableConfig(propertyStore, TableNameBuilder.OFFLINE.tableNameWithType(tableName), replaceVariables);
+  }
+
+  /**
+   * Get the realtime table config for the given table name. Any environment variables and system properties will be
+   * replaced with their actual values.
+   *
+   * @param tableName Table name with or without type suffix
+   * @return Table config
+   */
   @Nullable
   public static TableConfig getRealtimeTableConfig(ZkHelixPropertyStore<ZNRecord> propertyStore, String tableName) {
-    return getTableConfig(propertyStore, TableNameBuilder.REALTIME.tableNameWithType(tableName));
+    return getRealtimeTableConfig(propertyStore, tableName, true);
+  }
+
+  /**
+   * Get the realtime table config for the given table name.
+   *
+   * @param tableName Table name with or without type suffix
+   * @param replaceVariables Whether to replace environment variables and system properties with their actual values
+   * @return Table config
+   */
+  @Nullable
+  public static TableConfig getRealtimeTableConfig(ZkHelixPropertyStore<ZNRecord> propertyStore, String tableName,
+      boolean replaceVariables) {
+    return getTableConfig(propertyStore, TableNameBuilder.REALTIME.tableNameWithType(tableName), replaceVariables);
   }
 
   public static List<TableConfig> getAllTableConfigs(ZkHelixPropertyStore<ZNRecord> propertyStore) {
@@ -502,12 +559,17 @@ public static List<TableConfig> getAllTableConfigs(ZkHelixPropertyStore<ZNRecord
 
   @Nullable
   private static TableConfig toTableConfig(@Nullable ZNRecord znRecord) {
+    return toTableConfig(znRecord, true);
+  }
+
+  @Nullable
+  private static TableConfig toTableConfig(@Nullable ZNRecord znRecord, boolean replaceVariables) {
     if (znRecord == null) {
       return null;
     }
     try {
       TableConfig tableConfig = TableConfigUtils.fromZNRecord(znRecord);
-      return ConfigUtils.applyConfigWithEnvVariablesAndSystemProperties(tableConfig);
+      return replaceVariables ? ConfigUtils.applyConfigWithEnvVariablesAndSystemProperties(tableConfig) : tableConfig;
     } catch (Exception e) {
       LOGGER.error("Caught exception while creating table config from ZNRecord: {}", znRecord.getId(), e);
       return null;

pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/PinotTableRestletResource.java

@@ -378,14 +378,14 @@ public String listTableConfigs(
 
       if ((tableTypeStr == null || TableType.OFFLINE.name().equalsIgnoreCase(tableTypeStr))
           && _pinotHelixResourceManager.hasOfflineTable(tableName)) {
-        TableConfig tableConfig = _pinotHelixResourceManager.getOfflineTableConfig(tableName);
+        TableConfig tableConfig = _pinotHelixResourceManager.getOfflineTableConfig(tableName, false);
         Preconditions.checkNotNull(tableConfig);
         ret.set(TableType.OFFLINE.name(), tableConfig.toJsonNode());
       }
 
       if ((tableTypeStr == null || TableType.REALTIME.name().equalsIgnoreCase(tableTypeStr))
           && _pinotHelixResourceManager.hasRealtimeTable(tableName)) {
-        TableConfig tableConfig = _pinotHelixResourceManager.getRealtimeTableConfig(tableName);
+        TableConfig tableConfig = _pinotHelixResourceManager.getRealtimeTableConfig(tableName, false);
         Preconditions.checkNotNull(tableConfig);
         ret.set(TableType.REALTIME.name(), tableConfig.toJsonNode());
       }

pinot-controller/src/main/java/org/apache/pinot/controller/api/resources/TableConfigsRestletResource.java

@@ -156,8 +156,8 @@ public String getConfig(
     try {
       tableName = DatabaseUtils.translateTableName(tableName, headers);
       Schema schema = _pinotHelixResourceManager.getTableSchema(tableName);
-      TableConfig offlineTableConfig = _pinotHelixResourceManager.getOfflineTableConfig(tableName);
-      TableConfig realtimeTableConfig = _pinotHelixResourceManager.getRealtimeTableConfig(tableName);
+      TableConfig offlineTableConfig = _pinotHelixResourceManager.getOfflineTableConfig(tableName, false);
+      TableConfig realtimeTableConfig = _pinotHelixResourceManager.getRealtimeTableConfig(tableName, false);
       TableConfigs config = new TableConfigs(tableName, schema, offlineTableConfig, realtimeTableConfig);
       return config.toJsonString();
     } catch (Exception e) {

pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/PinotHelixResourceManager.java

@@ -3093,25 +3093,51 @@ public List<TableConfig> getAllTableConfigs() {
   }
 
   /**
-   * Get the offline table config for the given table name.
+   * Get the offline table config for the given table name. Any environment variables and system properties will be
+   * replaced with their actual values.
    *
    * @param tableName Table name with or without type suffix
    * @return Table config
    */
   @Nullable
   public TableConfig getOfflineTableConfig(String tableName) {
-    return ZKMetadataProvider.getOfflineTableConfig(_propertyStore, tableName);
+    return getOfflineTableConfig(tableName, true);
   }
 
   /**
-   * Get the realtime table config for the given table name.
+   * Get the offline table config for the given table name.
+   *
+   * @param tableName Table name with or without type suffix
+   * @param replaceVariables Whether to replace environment variables and system properties with their actual values
+   * @return Table config
+   */
+  @Nullable
+  public TableConfig getOfflineTableConfig(String tableName, boolean replaceVariables) {
+    return ZKMetadataProvider.getOfflineTableConfig(_propertyStore, tableName, replaceVariables);
+  }
+
+  /**
+   * Get the realtime table config for the given table name. Any environment variables and system properties will be
+   * replaced with their actual values.
    *
    * @param tableName Table name with or without type suffix
    * @return Table config
    */
   @Nullable
   public TableConfig getRealtimeTableConfig(String tableName) {
-    return ZKMetadataProvider.getRealtimeTableConfig(_propertyStore, tableName);
+    return getRealtimeTableConfig(tableName, true);
+  }
+
+  /**
+   * Get the realtime table config for the given table name.
+   *
+   * @param tableName Table name with or without type suffix
+   * @param replaceVariables Whether to replace environment variables and system properties with their actual values
+   * @return Table config
+   */
+  @Nullable
+  public TableConfig getRealtimeTableConfig(String tableName, boolean replaceVariables) {
+    return ZKMetadataProvider.getRealtimeTableConfig(_propertyStore, tableName, replaceVariables);
   }
 
   /**
@@ -3438,8 +3464,8 @@ public List<String> getOnlineUnTaggedBrokerInstanceList() {
    * @return List of untagged online server instances.
    */
   public List<String> getOnlineUnTaggedServerInstanceList() {
-    List<String> instanceListWithoutTags = HelixHelper.getInstancesWithoutTag(_helixZkManager,
-        Helix.UNTAGGED_SERVER_INSTANCE);
+    List<String> instanceListWithoutTags =
+        HelixHelper.getInstancesWithoutTag(_helixZkManager, Helix.UNTAGGED_SERVER_INSTANCE);
     List<String> liveInstances = _helixDataAccessor.getChildNames(_keyBuilder.liveInstances());
 
     instanceListWithoutTags.retainAll(liveInstances);