Add a flag to check the new metadata api is called for DTS users only

src/main/java/io/cdap/plugin/servicenow/ServiceNowBaseConfig.java

@@ -123,7 +123,7 @@ public void validateTable(String tableName, SourceValueType valueType, FailureCo
                             String tableField) {
     // Call API to fetch first record from the table
     ServiceNowTableAPIRequestBuilder requestBuilder = new ServiceNowTableAPIRequestBuilder(
-      connection.getRestApiEndpoint(), tableName, false)
+      connection.getRestApiEndpoint(), tableName, false, false)
       .setExcludeReferenceLink(true)
       .setDisplayValue(valueType)
       .setLimit(1);

src/main/java/io/cdap/plugin/servicenow/apiclient/ServiceNowTableAPIClientImpl.java

@@ -34,6 +34,8 @@
 import io.cdap.plugin.servicenow.restapi.RestAPIResponse;
 import io.cdap.plugin.servicenow.sink.model.APIResponse;
 import io.cdap.plugin.servicenow.sink.model.CreateRecordAPIResponse;
+import io.cdap.plugin.servicenow.sink.model.MetadataAPISchemaResponse;
+import io.cdap.plugin.servicenow.sink.model.SchemaField;
 import io.cdap.plugin.servicenow.sink.model.SchemaResponse;
 import io.cdap.plugin.servicenow.sink.model.ServiceNowSchemaField;
 import io.cdap.plugin.servicenow.util.SchemaBuilder;
@@ -127,7 +129,7 @@ public List<Map<String, String>> fetchTableRecords(
       int limit)
       throws ServiceNowAPIException {
     ServiceNowTableAPIRequestBuilder requestBuilder = new ServiceNowTableAPIRequestBuilder(
-      this.conf.getRestApiEndpoint(), tableName, false)
+      this.conf.getRestApiEndpoint(), tableName, false, false)
       .setExcludeReferenceLink(true)
       .setDisplayValue(valueType)
       .setLimit(limit);
@@ -265,7 +267,11 @@ public Schema fetchTableSchema(String tableName, FailureCollector collector) {
   }
 
   @VisibleForTesting
-  public SchemaResponse parseSchemaResponse(String responseBody) {
+  public MetadataAPISchemaResponse parseSchemaResponse(String responseBody) {
+    return GSON.fromJson(responseBody, MetadataAPISchemaResponse.class);
+  }
+
+  public SchemaResponse parseSchemaResponseWithoutMetadata(String responseBody) {
     return GSON.fromJson(responseBody, SchemaResponse.class);
   }
 
@@ -278,7 +284,7 @@ public SchemaResponse parseSchemaResponse(String responseBody) {
    */
   public Schema fetchTableSchema(String tableName, SourceValueType valueType)
       throws ServiceNowAPIException {
-      return fetchTableSchema(tableName, getAccessToken(), valueType);
+      return fetchTableSchema(tableName, getAccessToken(), valueType, false);
   }
 
   /**
@@ -287,25 +293,86 @@ public Schema fetchTableSchema(String tableName, SourceValueType valueType)
    * @param tableName ServiceNow table name for which schema is getting fetched
    * @param accessToken Access Token to use
    * @param valueType Type of value (Actual/Display)
+   * @param isMetadataAPIRequired Flag to determine whether the schema should be fetched via the Metadata API.
    * @return schema for given ServiceNow table
    */
-  public Schema fetchTableSchema(String tableName, String accessToken, SourceValueType valueType)
+  public Schema fetchTableSchema(String tableName, String accessToken, SourceValueType valueType,
+                                 boolean isMetadataAPIRequired)
       throws ServiceNowAPIException {
     ServiceNowTableAPIRequestBuilder requestBuilder = new ServiceNowTableAPIRequestBuilder(
-      this.conf.getRestApiEndpoint(), tableName, true)
+      this.conf.getRestApiEndpoint(), tableName, true, isMetadataAPIRequired)
       .setExcludeReferenceLink(true);
 
     RestAPIResponse restAPIResponse;
     requestBuilder.setAuthHeader(accessToken);
     restAPIResponse = executeGetWithRetries(requestBuilder.build());
-    SchemaResponse schemaResponse = parseSchemaResponse(restAPIResponse.getResponseBody());
     List<ServiceNowColumn> columns = new ArrayList<>();
 
-    if (schemaResponse.getResult() == null && schemaResponse.getResult().getColumns().isEmpty()) {
+    if (isMetadataAPIRequired) {
+      return prepareSchemaWithMetadataAPI(restAPIResponse, columns, tableName, valueType);
+    } else {
+      return prepareSchemaWithSchemaAPI(restAPIResponse, columns, tableName);
+    }
+  }
+
+  /**
+   * Processes a schema response obtained from the ServiceNow Table API (without using Metadata API)
+   * and constructs a {@link Schema} object based on the parsed column definitions.
+   *
+   * <p>This method parses the raw JSON response body into a list of schema fields,
+   * extracts the internal column types, and appends them to the provided column list.
+   * The final schema is constructed using the {@link SchemaBuilder} utility.</p>
+   *
+   * @param restAPIResponse The raw API response received from the ServiceNow Table API.
+   * @param columns         A list to which parsed {@link ServiceNowColumn} objects will be added.
+   * @param tableName       The name of the table for which the schema is being constructed.
+   *
+   * @return A {@link Schema} object representing the table structure as interpreted from the Schema API.
+   *
+   * @throws RuntimeException if the schema response is null or contains no result.
+   */
+  private Schema prepareSchemaWithSchemaAPI(RestAPIResponse restAPIResponse, List<ServiceNowColumn> columns,
+                                            String tableName) {
+    SchemaResponse schemaResponse =
+      parseSchemaResponseWithoutMetadata(restAPIResponse.getResponseBody());
+
+    if (schemaResponse.getResult() == null && schemaResponse.getResult().isEmpty()) {
+      throw new RuntimeException("Error - Schema Response does not contain any result");
+    }
+
+    for (SchemaField field : schemaResponse.getResult()) {
+      columns.add(new ServiceNowColumn(field.getName(), field.getInternalType()));
+    }
+    return SchemaBuilder.constructSchema(tableName, columns);
+  }
+
+  /**
+   * Parses a ServiceNow schema response obtained via the Metadata API and constructs a
+   * {@link Schema} object using the extracted field information and value type preferences.
+   *
+   * <p>This method reads the JSON response, extracts the column metadata including field names
+   * and data types, and adds each as a {@link ServiceNowColumn} to the provided list. The choice
+   * between internal values and display values is based on the {@code valueType} parameter.</p>
+   *
+   * @param restAPIResponse The response returned from the ServiceNow Metadata API.
+   * @param columns         A list to which parsed {@link ServiceNowColumn} definitions will be added.
+   * @param tableName       The name of the ServiceNow table for which the schema is being generated.
+   * @param valueType       The value type preference (e.g., {@code SHOW_DISPLAY_VALUE} or {@code USE_INTERNAL_VALUE}).
+   *                        Determines whether to use display types or internal types in the resulting schema.
+   *
+   * @return A {@link Schema} object representing the table structure as interpreted from the Metadata API.
+   *
+   * @throws RuntimeException if the response does not contain valid column information.
+   */
+  private Schema prepareSchemaWithMetadataAPI(RestAPIResponse restAPIResponse, List<ServiceNowColumn> columns,
+                                              String tableName, SourceValueType valueType) {
+    MetadataAPISchemaResponse metadataAPISchemaResponse = parseSchemaResponse(restAPIResponse.getResponseBody());
+
+    if (metadataAPISchemaResponse.getResult() == null && metadataAPISchemaResponse.getResult().getColumns().isEmpty()) {
       throw new RuntimeException("Error - Schema Response does not contain any result");
     }
 
-    for (ServiceNowSchemaField field : schemaResponse.getResult().getColumns().values()) {
+    for (ServiceNowSchemaField field : metadataAPISchemaResponse.getResult().getColumns().values()) {
       if (valueType.equals(SourceValueType.SHOW_DISPLAY_VALUE) &&
         !Objects.equals(field.getType(), field.getInternalType())) {
         columns.add(new ServiceNowColumn(field.getName(), field.getType()));
@@ -338,7 +405,7 @@ public int getTableRecordCount(String tableName)
    */
   public int getTableRecordCount(String tableName, String accessToken) throws ServiceNowAPIException {
     ServiceNowTableAPIRequestBuilder requestBuilder = new ServiceNowTableAPIRequestBuilder(
-      this.conf.getRestApiEndpoint(), tableName, false)
+      this.conf.getRestApiEndpoint(), tableName, false, false)
       .setExcludeReferenceLink(true)
       .setDisplayValue(SourceValueType.SHOW_DISPLAY_VALUE)
       .setLimit(1);
@@ -358,7 +425,7 @@ public int getTableRecordCount(String tableName, String accessToken) throws Serv
    */
   public String createRecord(String tableName, HttpEntity entity) throws IOException, ServiceNowAPIException {
     ServiceNowTableAPIRequestBuilder requestBuilder = new ServiceNowTableAPIRequestBuilder(
-      this.conf.getRestApiEndpoint(), tableName, false);
+      this.conf.getRestApiEndpoint(), tableName, false, false);
     String systemID;
     RestAPIResponse apiResponse = null;
     try {
@@ -393,7 +460,7 @@ public Map<String, String> getRecordFromServiceNowTable(String tableName, String
       throws ServiceNowAPIException {
 
     ServiceNowTableAPIRequestBuilder requestBuilder = new ServiceNowTableAPIRequestBuilder(
-      this.conf.getRestApiEndpoint(), tableName, false)
+      this.conf.getRestApiEndpoint(), tableName, false, false)
       .setQuery(query);
 
     RestAPIResponse restAPIResponse;

src/main/java/io/cdap/plugin/servicenow/apiclient/ServiceNowTableAPIRequestBuilder.java

@@ -48,9 +48,14 @@ public class ServiceNowTableAPIRequestBuilder extends RestAPIRequest.Builder {
    */
   private static final String METADATA_API_URL_TEMPLATE = "%s/api/now/ui/meta/%s";
 
-  public ServiceNowTableAPIRequestBuilder(String instanceBaseUrl, String tableName, boolean isSchemaRequired) {
+  public ServiceNowTableAPIRequestBuilder(String instanceBaseUrl, String tableName, boolean isSchemaRequired,
+                                          boolean isMetadataAPIRequired) {
     if (isSchemaRequired) {
-      this.setUrl(String.format(METADATA_API_URL_TEMPLATE, instanceBaseUrl, tableName));
+      if (isMetadataAPIRequired) {
+        this.setUrl(String.format(METADATA_API_URL_TEMPLATE, instanceBaseUrl, tableName));
+      } else {
+        this.setUrl(String.format(SCHEMA_API_URL_TEMPLATE, instanceBaseUrl, tableName));
+      }
     } else {
       this.setUrl(String.format(TABLE_API_URL_TEMPLATE, instanceBaseUrl, tableName));
     }

src/main/java/io/cdap/plugin/servicenow/connector/ServiceNowConnector.java

@@ -127,7 +127,7 @@ public BrowseDetail browse(ConnectorContext connectorContext,
    */
   private TableList listTables(String accessToken) throws ServiceNowAPIException {
     ServiceNowTableAPIRequestBuilder requestBuilder = new ServiceNowTableAPIRequestBuilder(
-      config.getRestApiEndpoint(), OBJECT_TABLE_LIST, false);
+      config.getRestApiEndpoint(), OBJECT_TABLE_LIST, false, false);
     requestBuilder.setAuthHeader(accessToken);
     requestBuilder.setAcceptHeader(MediaType.APPLICATION_JSON);
     requestBuilder.setContentTypeHeader(MediaType.APPLICATION_JSON);
@@ -173,7 +173,7 @@ public List<StructuredRecord> sample(ConnectorContext connectorContext, SampleRe
   private List<StructuredRecord> getTableData(String tableName, int limit)
       throws OAuthProblemException, OAuthSystemException, ServiceNowAPIException {
     ServiceNowTableAPIRequestBuilder requestBuilder = new ServiceNowTableAPIRequestBuilder(
-      config.getRestApiEndpoint(), tableName, false)
+      config.getRestApiEndpoint(), tableName, false, false)
       .setExcludeReferenceLink(true)
       .setDisplayValue(SourceValueType.SHOW_DISPLAY_VALUE)
       .setLimit(limit);

src/main/java/io/cdap/plugin/servicenow/sink/model/MetadataAPISchemaResponse.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright © 2022 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.plugin.servicenow.sink.model;
+
+/**
+ * Model class for Schema Response from Column Metadata API
+ */
+public class MetadataAPISchemaResponse {
+
+  private final MetadataAPISchemaResult result;
+
+  public MetadataAPISchemaResponse(MetadataAPISchemaResult result) {
+    this.result = result;
+  }
+
+  public MetadataAPISchemaResult getResult() {
+    return result;
+  }
+
+}

src/main/java/io/cdap/plugin/servicenow/sink/model/ServiceNowSchemaResult.java → src/main/java/io/cdap/plugin/servicenow/sink/model/MetadataAPISchemaResult.java

@@ -50,10 +50,10 @@
  * In this example, the map will contain keys like {@code "state"} and {@code "active"},
  * each pointing to a {@code ServiceNowSchemaField} instance with metadata about that field.
  */
-public class ServiceNowSchemaResult {
+public class MetadataAPISchemaResult {
   private final Map<String, ServiceNowSchemaField> columns;
 
-  public ServiceNowSchemaResult(Map<String, ServiceNowSchemaField> columns) {
+  public MetadataAPISchemaResult(Map<String, ServiceNowSchemaField> columns) {
     this.columns = columns;
   }
 

src/main/java/io/cdap/plugin/servicenow/sink/model/SchemaField.java

@@ -0,0 +1,50 @@
+/*
+ * Copyright © 2025 Cask Data, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not
+ * use this file except in compliance with the License. You may obtain a copy of
+ * the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.plugin.servicenow.sink.model;
+
+/**
+ * Model class for Schema Field from Schema API
+ */
+public class SchemaField {
+  private final String label;
+  private final String exampleValue;
+  private final String internalType;
+  private final String name;
+
+  public SchemaField(String label, String exampleValue, String internalType, String name) {
+    this.label = label;
+    this.exampleValue = exampleValue;
+    this.internalType = internalType;
+    this.name = name;
+  }
+
+  public String getLabel() {
+    return label;
+  }
+
+  public String getExampleValue() {
+    return exampleValue;
+  }
+
+  public String getInternalType() {
+    return internalType;
+  }
+
+  public String getName() {
+    return name;
+  }
+}

src/main/java/io/cdap/plugin/servicenow/sink/model/SchemaResponse.java

@@ -1,5 +1,5 @@
 /*
- * Copyright © 2022 Cask Data, Inc.
+ * Copyright © 2025 Cask Data, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License"); you may not
  * use this file except in compliance with the License. You may obtain a copy of
@@ -16,19 +16,21 @@
 
 package io.cdap.plugin.servicenow.sink.model;
 
+import java.util.List;
+
 /**
  * Model class for Schema Response from Schema API
  */
 public class SchemaResponse {
 
-  private final ServiceNowSchemaResult result;
+  private final List<SchemaField> result;
 
-  public SchemaResponse(ServiceNowSchemaResult result) {
+  public SchemaResponse(List<SchemaField> result) {
     this.result = result;
   }
 
-  public ServiceNowSchemaResult getResult() {
+  public List<SchemaField> getResult() {
     return result;
   }
-  
+
 }

src/main/java/io/cdap/plugin/servicenow/sink/model/ServiceNowSchemaField.java

@@ -19,7 +19,7 @@
 import com.google.gson.annotations.SerializedName;
 
 /**
- * Model class for Schema Field from Schema API
+ * Model class for Schema Field from Column Metadata API
  */
 public class ServiceNowSchemaField {
   private final String label;