Fix IndexOutOfBoundsException in SalesforceWideRecordReader when bulk…

src/main/java/io/cdap/plugin/salesforce/authenticator/Authenticator.java

@@ -18,24 +18,36 @@
 
 import com.google.common.base.Strings;
 import com.google.gson.Gson;
+import com.google.gson.JsonSyntaxException;
 import com.sforce.ws.ConnectorConfig;
 import io.cdap.plugin.salesforce.SalesforceConnectionUtil;
 import io.cdap.plugin.salesforce.SalesforceConstants;
 import io.cdap.plugin.salesforce.plugin.OAuthInfo;
 import org.eclipse.jetty.client.HttpClient;
 import org.eclipse.jetty.client.HttpProxy;
 import org.eclipse.jetty.client.ProxyConfiguration;
+import org.eclipse.jetty.client.api.ContentResponse;
 import org.eclipse.jetty.client.api.Request;
 import org.eclipse.jetty.util.ssl.SslContextFactory;
 
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 /**
  * Authentication to Salesforce via oauth2
  */
 public class Authenticator {
   private static final Gson GSON = new Gson();
+  /**
+   * Matches an HTML {@code <div ... id="error" ...>...</div>} element returned
+   * by Salesforce HTML login/error pages instead of JSON, capturing its inner
+   * content across multiple lines (DOTALL) regardless of tag case.
+   */
+  private static final Pattern HTML_ERROR_PATTERN =
+      Pattern.compile("<div\\b[^>]*\\bid=['\"]?error['\"]?[^>]*>(.*?)</div>",
+          Pattern.DOTALL | Pattern.CASE_INSENSITIVE);
 
   /**
    * Authenticates via oauth2 to salesforce and returns a connectorConfig
@@ -114,8 +126,30 @@ public static OAuthInfo getOAuthInfo(AuthenticatorCredentials credentials) throw
           break;
       }
 
-      String response = authRequest.send().getContentAsString();
-      AuthResponse authResponse = GSON.fromJson(response, AuthResponse.class);
+      ContentResponse httpResponse = authRequest.send();
+      int statusCode = httpResponse.getStatus();
+      String reason = httpResponse.getReason();
+      String response = httpResponse.getContentAsString();
+      AuthResponse authResponse;
+
+      try {
+        authResponse = GSON.fromJson(response, AuthResponse.class);
+      } catch (JsonSyntaxException e) {
+        throw new IllegalStateException(
+            String.format(
+                "Cannot authenticate to Salesforce (HTTP %d %s). %s",
+                statusCode,
+                Strings.nullToEmpty(reason),
+                extractErrorFromResponse(response)),
+            e);
+      }
+
+      if (authResponse == null) {
+        throw new IllegalStateException(
+            String.format(
+                "Empty OAuth response from Salesforce (HTTP %d %s).",
+                statusCode, Strings.nullToEmpty(reason)));
+      }
 
       if (!Strings.isNullOrEmpty(authResponse.getError())) {
         throw new IllegalArgumentException(
@@ -150,4 +184,43 @@ public static void setProxy(AuthenticatorCredentials credentials, HttpClient htt
                                                          "given proxy server details. Error : %s", e.getMessage()), e);
     }
   }
+
+  /**
+   * Extracts a concise, human-readable error message from a non-JSON Salesforce
+   * OAuth response instead of embedding the entire raw HTML page in the
+   * exception message.
+   *
+   * <p>It searches the response for an HTML error container matching
+   * {@link #HTML_ERROR_PATTERN}, strips any nested HTML tags from its inner
+   * content, and trims surrounding whitespace. If no non-empty error container
+   * is found, it returns a generic authentication error message.
+   *
+   * <p>Example:
+   * <pre>{@code
+   * // 1. Input HTML containing an id="error" div:
+   * "<div class="loginError" id="error">Error: <span>Enter username.</span></div>"
+   * // Returns: "Error: Enter username."
+   *
+   * // 2. Input HTML without an id="error" div (e.g., 503 maintenance page):
+   * "<html><body><h1>503 Service Unavailable</h1></body></html>"
+   * // Returns: "Error while authenticating to Salesforce with given credentials."
+   * }</pre>
+   *
+   * @param response raw HTTP response body returned by Salesforce OAuth
+   * @return extracted plain-text error message, or a default error message
+   */
+  private static String extractErrorFromResponse(String response) {
+    if (!Strings.isNullOrEmpty(response)) {
+      Matcher matcher = HTML_ERROR_PATTERN.matcher(response);
+      if (matcher.find()) {
+        // Strip any nested HTML tags inside the <div id="error">...</div> block (e.g.,
+        // <span>, <b>).
+        String error = matcher.group(1).replaceAll("<[^>]+>", "").trim();
+        if (!error.isEmpty()) {
+          return error;
+        }
+      }
+    }
+    return "Error while authenticating to Salesforce with given credentials.";
+  }
 }

src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceWideRecordReader.java

@@ -34,6 +34,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -106,8 +107,11 @@ public SalesforceWideRecordReader initialize(
 
     batchIterator = partitions.iterator();
 
-    // initialize the 1st batch
-    results = fetchBatchRecords();
+    if (batchIterator.hasNext()) {
+      results = fetchBatchRecords();
+    } else {
+      results = Collections.emptyList();
+    }
 
     return this;
   }

src/test/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceWideRecordReaderTest.java

@@ -16,11 +16,17 @@
 
 package io.cdap.plugin.salesforce.plugin.source.batch;
 
+import com.sun.net.httpserver.HttpServer;
 import io.cdap.cdap.api.data.schema.Schema;
+import io.cdap.plugin.salesforce.authenticator.AuthenticatorCredentials;
+import io.cdap.plugin.salesforce.plugin.OAuthInfo;
 import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
 import org.junit.Test;
 import org.mockito.internal.util.reflection.FieldSetter;
+
+import java.net.InetSocketAddress;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -31,6 +37,8 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doNothing;
@@ -106,4 +114,43 @@ public void testWideRecordReader_fetchBatchRecordsCalledExpectedNumberOfTimes()
     verify(reader, times(4)).fetchBatchRecords();
     assertEquals(3, readRecords.size());
   }
+
+  @Test
+  public void nextKeyValue_withEmptyBatch_returnsFalse() throws Exception {
+    HttpServer server = HttpServer.create(new InetSocketAddress(0), 0);
+    server.createContext("/services/async/", exchange -> {
+      String response = exchange.getRequestURI().getPath().endsWith("/result")
+        ? "<result-list xmlns=\"http://www.force.com/2009/06/asyncapi/dataload\"><result>res1</result></result-list>"
+        : "<batchInfo xmlns=\"http://www.force.com/2009/06/asyncapi/dataload\">"
+            + "<id>batch1</id><jobId>job1</jobId><state>Completed</state></batchInfo>";
+      byte[] bytes = response.getBytes(StandardCharsets.UTF_8);
+      exchange.getResponseHeaders().add("Content-Type", "application/xml");
+      exchange.sendResponseHeaders(200, bytes.length);
+      exchange.getResponseBody().write(bytes);
+      exchange.close();
+    });
+    server.start();
+
+    try {
+      Schema schema = Schema.recordOf("test", Schema.Field.of("Id", Schema.of(Schema.Type.STRING)));
+      SoapRecordToMapTransformer transformer = mock(SoapRecordToMapTransformer.class);
+      SalesforceWideRecordReader reader = spy(new SalesforceWideRecordReader(
+        schema, "SELECT Id FROM Account", transformer));
+      doNothing().when(reader).setupParser();
+      SalesforceSplit split = new SalesforceSplit("job1", "batch1", "SELECT Id FROM Account");
+      OAuthInfo oAuthInfo = new OAuthInfo("token", "http://localhost:" + server.getAddress().getPort());
+      AuthenticatorCredentials credentials = AuthenticatorCredentials.fromParameters(
+        oAuthInfo, 5000, 5000, null, 1L, 1L, 1, false);
+
+      reader.initialize(split, credentials);
+      boolean hasNext = reader.nextKeyValue();
+
+      assertFalse(hasNext);
+      assertEquals(0.0f, reader.getProgress(), 0.0001);
+      assertNull(reader.getCurrentValue());
+      verify(reader, times(0)).fetchBatchRecords();
+    } finally {
+      server.stop(0);
+    }
+  }
 }