...
} 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("]*\\bid=['\"]?error['\"]?[^>]*>(.*?)",
+ 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.
+ *
+ * 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.
+ *
+ *
Example:
+ *
{@code
+ * // 1. Input HTML containing an id="error" div:
+ * "Error: Enter username.
"
+ * // Returns: "Error: Enter username."
+ *
+ * // 2. Input HTML without an id="error" div (e.g., 503 maintenance page):
+ * "503 Service Unavailable
"
+ * // Returns: "Error while authenticating to Salesforce with given credentials."
+ * }
+ *
+ * @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 ...
block (e.g.,
+ // , ).
+ String error = matcher.group(1).replaceAll("<[^>]+>", "").trim();
+ if (!error.isEmpty()) {
+ return error;
+ }
+ }
+ }
+ return "Error while authenticating to Salesforce with given credentials.";
+ }
}
diff --git a/src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceWideRecordReader.java b/src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceWideRecordReader.java
index 2299fa6d..2d69c3eb 100644
--- a/src/main/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceWideRecordReader.java
+++ b/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;
}
diff --git a/src/test/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceWideRecordReaderTest.java b/src/test/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceWideRecordReaderTest.java
index 35a24c38..a3021783 100644
--- a/src/test/java/io/cdap/plugin/salesforce/plugin/source/batch/SalesforceWideRecordReaderTest.java
+++ b/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")
+ ? "res1"
+ : ""
+ + "batch1job1Completed";
+ 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);
+ }
+ }
}