();
+ for (ETDataExtensionColumn column : columns) {
+ columnNames.add(column.getName());
+ }
+ return columnNames;
+ }
+
+ @Override
+ public String getId() {
+ return null;
+ }
+
+ @Override
+ public void setId(String s) {
+
+ }
+}
diff --git a/src/main/java/com/custom/fuelsdk/internal/ClickEvent.java b/src/main/java/com/custom/fuelsdk/internal/ClickEvent.java
new file mode 100644
index 0000000..19b0b7f
--- /dev/null
+++ b/src/main/java/com/custom/fuelsdk/internal/ClickEvent.java
@@ -0,0 +1,133 @@
+/*
+ * Copyright © 2020 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 com.custom.fuelsdk.internal;
+
+import com.exacttarget.fuelsdk.internal.TrackingEvent;
+import org.apache.commons.lang.builder.ToStringBuilder;
+import org.apache.cxf.xjc.runtime.JAXBToStringStyle;
+
+import javax.xml.bind.annotation.XmlAccessType;
+import javax.xml.bind.annotation.XmlAccessorType;
+import javax.xml.bind.annotation.XmlElement;
+import javax.xml.bind.annotation.XmlType;
+
+
+/**
+ * Java class for ClickEvent complex type.
+ *
+ *
The following schema fragment specifies the expected content contained within this class.
+ *
+ *
+ * <complexType name="ClickEvent">
+ * <complexContent>
+ * <extension base="{http://exacttarget.com/wsdl/partnerAPI}TrackingEvent">
+ * <sequence>
+ * <element name="URLID" type="{http://www.w3.org/2001/XMLSchema}int" minOccurs="0"/>
+ * <element name="URL" type="{http://www.w3.org/2001/XMLSchema}string" minOccurs="0"/>
+ * <element name="URLIDLong" type="{http://www.w3.org/2001/XMLSchema}long" minOccurs="0"/>
+ * </sequence>
+ * </extension>
+ * </complexContent>
+ * </complexType>
+ *
+ */
+@XmlAccessorType(XmlAccessType.FIELD)
+@XmlType(name = "ClickEvent", propOrder = {
+ "urlid",
+ "url",
+ "urlIdLong"
+})
+public class ClickEvent
+ extends TrackingEvent {
+
+ @XmlElement(name = "URLID")
+ protected Integer urlid;
+ @XmlElement(name = "URL")
+ protected String url;
+ @XmlElement(name = "URLIDLong")
+ protected Long urlIdLong;
+
+ /**
+ * Gets the value of the urlid property.
+ *
+ * @return possible object is
+ * {@link Integer }
+ */
+ public Integer getUrlid() {
+ return urlid;
+ }
+
+ /**
+ * Sets the value of the urlid property.
+ *
+ * @param value allowed object is
+ * {@link Integer }
+ */
+ public void setUrlid(Integer value) {
+ this.urlid = value;
+ }
+
+ /**
+ * Gets the value of the url property.
+ *
+ * @return possible object is
+ * {@link String }
+ */
+ public String getUrl() {
+ return url;
+ }
+
+ /**
+ * Sets the value of the url property.
+ *
+ * @param value allowed object is
+ * {@link String }
+ */
+ public void setUrl(String value) {
+ this.url = value;
+ }
+
+ /**
+ * Gets the value of the urlIdLong property.
+ *
+ * @return possible object is
+ * {@link Long }
+ */
+ public Long getUrlIdLong() {
+ return urlIdLong;
+ }
+
+ /**
+ * Sets the value of the urlIdLong property.
+ *
+ * @param value allowed object is
+ * {@link Long }
+ */
+ public void setUrlIdLong(Long value) {
+ this.urlIdLong = value;
+ }
+
+ /**
+ * Generates a String representation of the contents of this type.
+ * This is an extension method, produced by the 'ts' xjc plugin
+ */
+ @Override
+ public String toString() {
+ return ToStringBuilder.reflectionToString(this, JAXBToStringStyle.DEFAULT_STYLE);
+ }
+
+}
diff --git a/src/main/java/io/cdap/plugin/sfmc/DataExtensionClient.java b/src/main/java/io/cdap/plugin/sfmc/sink/DataExtensionClient.java
similarity index 99%
rename from src/main/java/io/cdap/plugin/sfmc/DataExtensionClient.java
rename to src/main/java/io/cdap/plugin/sfmc/sink/DataExtensionClient.java
index f003396..303c08b 100644
--- a/src/main/java/io/cdap/plugin/sfmc/DataExtensionClient.java
+++ b/src/main/java/io/cdap/plugin/sfmc/sink/DataExtensionClient.java
@@ -14,7 +14,7 @@
* the License.
*/
-package io.cdap.plugin.sfmc;
+package io.cdap.plugin.sfmc.sink;
import com.exacttarget.fuelsdk.ETClient;
import com.exacttarget.fuelsdk.ETConfiguration;
diff --git a/src/main/java/io/cdap/plugin/sfmc/DataExtensionInfo.java b/src/main/java/io/cdap/plugin/sfmc/sink/DataExtensionInfo.java
similarity index 98%
rename from src/main/java/io/cdap/plugin/sfmc/DataExtensionInfo.java
rename to src/main/java/io/cdap/plugin/sfmc/sink/DataExtensionInfo.java
index 95f6e85..a551df8 100644
--- a/src/main/java/io/cdap/plugin/sfmc/DataExtensionInfo.java
+++ b/src/main/java/io/cdap/plugin/sfmc/sink/DataExtensionInfo.java
@@ -14,7 +14,7 @@
* the License.
*/
-package io.cdap.plugin.sfmc;
+package io.cdap.plugin.sfmc.sink;
import com.exacttarget.fuelsdk.ETDataExtensionColumn;
diff --git a/src/main/java/io/cdap/plugin/sfmc/DataExtensionOutputFormat.java b/src/main/java/io/cdap/plugin/sfmc/sink/DataExtensionOutputFormat.java
similarity index 99%
rename from src/main/java/io/cdap/plugin/sfmc/DataExtensionOutputFormat.java
rename to src/main/java/io/cdap/plugin/sfmc/sink/DataExtensionOutputFormat.java
index 34fbf72..7930b5f 100644
--- a/src/main/java/io/cdap/plugin/sfmc/DataExtensionOutputFormat.java
+++ b/src/main/java/io/cdap/plugin/sfmc/sink/DataExtensionOutputFormat.java
@@ -14,7 +14,7 @@
* the License.
*/
-package io.cdap.plugin.sfmc;
+package io.cdap.plugin.sfmc.sink;
import com.exacttarget.fuelsdk.ETSdkException;
import io.cdap.cdap.api.data.format.StructuredRecord;
diff --git a/src/main/java/io/cdap/plugin/sfmc/DataExtensionRecordWriter.java b/src/main/java/io/cdap/plugin/sfmc/sink/DataExtensionRecordWriter.java
similarity index 99%
rename from src/main/java/io/cdap/plugin/sfmc/DataExtensionRecordWriter.java
rename to src/main/java/io/cdap/plugin/sfmc/sink/DataExtensionRecordWriter.java
index 1052e1e..76226ef 100644
--- a/src/main/java/io/cdap/plugin/sfmc/DataExtensionRecordWriter.java
+++ b/src/main/java/io/cdap/plugin/sfmc/sink/DataExtensionRecordWriter.java
@@ -14,7 +14,7 @@
* the License.
*/
-package io.cdap.plugin.sfmc;
+package io.cdap.plugin.sfmc.sink;
import com.exacttarget.fuelsdk.ETDataExtensionRow;
import com.exacttarget.fuelsdk.ETResponse;
diff --git a/src/main/java/io/cdap/plugin/sfmc/MarketingCloudConf.java b/src/main/java/io/cdap/plugin/sfmc/sink/MarketingCloudConf.java
similarity index 99%
rename from src/main/java/io/cdap/plugin/sfmc/MarketingCloudConf.java
rename to src/main/java/io/cdap/plugin/sfmc/sink/MarketingCloudConf.java
index 7cb9bb7..a2ebedb 100644
--- a/src/main/java/io/cdap/plugin/sfmc/MarketingCloudConf.java
+++ b/src/main/java/io/cdap/plugin/sfmc/sink/MarketingCloudConf.java
@@ -14,7 +14,7 @@
* the License.
*/
-package io.cdap.plugin.sfmc;
+package io.cdap.plugin.sfmc.sink;
import com.exacttarget.fuelsdk.ETSdkException;
import io.cdap.cdap.api.annotation.Description;
diff --git a/src/main/java/io/cdap/plugin/sfmc/MarketingCloudDataExtensionSink.java b/src/main/java/io/cdap/plugin/sfmc/sink/MarketingCloudDataExtensionSink.java
similarity index 99%
rename from src/main/java/io/cdap/plugin/sfmc/MarketingCloudDataExtensionSink.java
rename to src/main/java/io/cdap/plugin/sfmc/sink/MarketingCloudDataExtensionSink.java
index a8936b6..31f0b3d 100644
--- a/src/main/java/io/cdap/plugin/sfmc/MarketingCloudDataExtensionSink.java
+++ b/src/main/java/io/cdap/plugin/sfmc/sink/MarketingCloudDataExtensionSink.java
@@ -14,7 +14,7 @@
* the License.
*/
-package io.cdap.plugin.sfmc;
+package io.cdap.plugin.sfmc.sink;
import io.cdap.cdap.api.annotation.Description;
import io.cdap.cdap.api.annotation.Name;
diff --git a/src/main/java/io/cdap/plugin/sfmc/Operation.java b/src/main/java/io/cdap/plugin/sfmc/sink/Operation.java
similarity index 95%
rename from src/main/java/io/cdap/plugin/sfmc/Operation.java
rename to src/main/java/io/cdap/plugin/sfmc/sink/Operation.java
index 72ce14e..417e1a5 100644
--- a/src/main/java/io/cdap/plugin/sfmc/Operation.java
+++ b/src/main/java/io/cdap/plugin/sfmc/sink/Operation.java
@@ -14,7 +14,7 @@
* the License.
*/
-package io.cdap.plugin.sfmc;
+package io.cdap.plugin.sfmc.sink;
/**
* The type of write operation.
diff --git a/src/main/java/io/cdap/plugin/sfmc/RecordDataExtensionRowConverter.java b/src/main/java/io/cdap/plugin/sfmc/sink/RecordDataExtensionRowConverter.java
similarity index 99%
rename from src/main/java/io/cdap/plugin/sfmc/RecordDataExtensionRowConverter.java
rename to src/main/java/io/cdap/plugin/sfmc/sink/RecordDataExtensionRowConverter.java
index e2e905d..0fefe6d 100644
--- a/src/main/java/io/cdap/plugin/sfmc/RecordDataExtensionRowConverter.java
+++ b/src/main/java/io/cdap/plugin/sfmc/sink/RecordDataExtensionRowConverter.java
@@ -14,7 +14,7 @@
* the License.
*/
-package io.cdap.plugin.sfmc;
+package io.cdap.plugin.sfmc.sink;
import com.exacttarget.fuelsdk.ETDataExtensionColumn;
import com.exacttarget.fuelsdk.ETDataExtensionRow;
diff --git a/src/main/java/io/cdap/plugin/sfmc/source/MarketingCloudClient.java b/src/main/java/io/cdap/plugin/sfmc/source/MarketingCloudClient.java
new file mode 100644
index 0000000..f1e6867
--- /dev/null
+++ b/src/main/java/io/cdap/plugin/sfmc/source/MarketingCloudClient.java
@@ -0,0 +1,241 @@
+/*
+ * Copyright © 2020 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.sfmc.source;
+
+//import com.exacttarget.fuelsdk.ETApiObject;
+
+import com.custom.fuelsdk.PaginationETSoapObject;
+import com.exacttarget.fuelsdk.ETClient;
+import com.exacttarget.fuelsdk.ETConfiguration;
+import com.exacttarget.fuelsdk.ETDataExtensionColumn;
+import com.exacttarget.fuelsdk.ETDataExtensionRow;
+import com.exacttarget.fuelsdk.ETExpression;
+import com.exacttarget.fuelsdk.ETFilter;
+import com.exacttarget.fuelsdk.ETResponse;
+import com.exacttarget.fuelsdk.ETSdkException;
+import com.exacttarget.fuelsdk.ETSoapObject;
+import io.cdap.plugin.sfmc.source.util.MarketingCloudColumn;
+import io.cdap.plugin.sfmc.source.util.MarketingCloudObjectInfo;
+import io.cdap.plugin.sfmc.source.util.SourceObject;
+import io.cdap.plugin.sfmc.source.util.Util;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * Utility class that provides methods to connect to Salesforce instance.
+ */
+public class MarketingCloudClient {
+ private final ETClient client;
+ private static final Logger LOG = LoggerFactory.getLogger(MarketingCloudClient.class);
+
+ MarketingCloudClient(ETClient client) {
+ this.client = client;
+ }
+
+ /**
+ * Initialize the connection with Salesforce Marketing Cloud using FuelSDK.
+ *
+ * @param clientId The Salesforce Marketing Cloud Client Id
+ * @param clientSecret The Salesforce Marketing Cloud Client Secret
+ * @param authEndpoint Auth Endpoint url for Salesforce Marketing Cloud
+ * @param soapEndpoint SOAP Endpoint url for Salesforce Marketing Cloud
+ * @return The instance of MarketingCloudClient object
+ * @throws ETSdkException The FuelSDKException
+ */
+ public static MarketingCloudClient create(String clientId, String clientSecret, String authEndpoint,
+ String soapEndpoint) throws ETSdkException {
+ ETConfiguration conf = new ETConfiguration();
+ conf.set("clientId", clientId);
+ conf.set("clientSecret", clientSecret);
+ conf.set("authEndpoint", authEndpoint);
+ conf.set("soapEndpoint", soapEndpoint);
+ conf.set("useOAuth2Authentication", "true");
+ ClassLoader oldCL = Thread.currentThread().getContextClassLoader();
+ try {
+ Thread.currentThread().setContextClassLoader(MarketingCloudClient.class.getClassLoader());
+ return new MarketingCloudClient(new ETClient(conf));
+ } finally {
+ Thread.currentThread().setContextClassLoader(oldCL);
+ }
+ }
+
+ /**
+ * Check if filter string is a valid SFMC Expression.
+ *
+ * @param filter
+ * @return
+ */
+ public static void validateFilter(String filter) throws ETSdkException {
+ if (!Util.isNullOrEmpty(filter)) {
+ ETExpression expression = ETExpression.parse(filter);
+ }
+ }
+
+ private ETExpression getExpressionfromString(String filter) throws ETSdkException {
+ if (!Util.isNullOrEmpty(filter)) {
+ return ETExpression.parse(filter);
+ } else {
+ return new ETExpression();
+ }
+ }
+
+ /**
+ * Fetch records for passed object from Salesforce Marketing Cloud.
+ *
+ * @param object The SourceObject which tells what data to be fetched from Salesforce Marketing Cloud
+ * @return The list of ETApiObject representing the records from requested object
+ */
+ public List fetchObjectRecords(SourceObject object) throws ETSdkException {
+ ETFilter filter = new ETFilter();
+ filter.setExpression(getExpressionfromString(object.getFilter()));
+ return fetchObjectData(client, object.getClassRef(), filter);
+ }
+
+ /**
+ * Fetch the schema information for passed object from Salesforce Marketing Cloud.
+ *
+ * @param object The SourceObject which tells for which schema to be fetched from Salesforce Marketing Cloud
+ * @return The instance of MarketingCloudObjectInfo object
+ * @throws ETSdkException The FuelSDKException
+ */
+ public MarketingCloudObjectInfo fetchObjectSchema(SourceObject object)
+ throws ETSdkException {
+ Class clazz = object.getClassRef();
+ return new MarketingCloudObjectInfo(object, fetchObjectFields(clazz));
+ }
+
+ /**
+ * Fetch records for passed object from Salesforce Marketing Cloud.
+ *
+ * @param dataExtensionKey The data extension key for which data to be fetched from Salesforce Marketing Cloud
+ * @return The list of ETDataExtensionRow representing the records from requested data extension
+ * @throws ETSdkException The FuelSDKException
+ */
+ public List fetchDataExtensionRecords(String dataExtensionKey, String filterStr)
+ throws ETSdkException {
+
+ ETFilter filter = new ETFilter();
+ filter.setExpression(getExpressionfromString(filterStr));
+
+ ETResponse response = PaginationETSoapObject.select(client, "key=" + dataExtensionKey,
+ filter);
+ List rows = response.getObjects();
+ while (response.getResponseMessage().equals("MoreDataAvailable")) {
+ response = PaginationETSoapObject.continueRequest(client, null, response.getRequestId(),
+ filter);
+ rows.addAll(response.getObjects());
+ }
+ return rows;
+ //return call(() -> ETDataExtension.select(client, "key=" + dataExtensionKey)
+ //.getObjects());
+ }
+
+ /**
+ * Fetch the schema information for passed object from Salesforce Marketing Cloud.
+ *
+ * @param dataExtensionKey The data extension key for which schema to be fetched from Salesforce Marketing Cloud
+ * @return The instance of MarketingCloudObjectInfo object
+ * @throws ETSdkException The FuelSDKException
+ */
+ public MarketingCloudObjectInfo fetchDataExtensionSchema(String dataExtensionKey)
+ throws ETSdkException {
+ return call(() -> {
+ ETExpression expression = buildDataExtensionExpression(dataExtensionKey);
+
+ ETFilter filter = new ETFilter();
+ filter.setExpression(expression);
+ filter.addProperty("name");
+ filter.addProperty("type");
+
+ ETResponse response = ETDataExtensionColumn.retrieve(client, ETDataExtensionColumn.class,
+ (Integer) null, (Integer) null,
+ filter);
+ List columns = response.getObjects().stream()
+ .map(o -> new MarketingCloudColumn(o.getName(), o.getType().name()))
+ .collect(Collectors.toList());
+
+ //Integer recordCount = fetchDataExtensionRecordCount(dataExtensionKey);
+
+ return new MarketingCloudObjectInfo(SourceObject.DATA_EXTENSION, dataExtensionKey, columns);
+ });
+ }
+
+ private ETExpression buildDataExtensionExpression(String dataExtensionKey) {
+ ETExpression expression = new ETExpression();
+ expression.setProperty("DataExtension.CustomerKey");
+ expression.setOperator(ETExpression.Operator.EQUALS);
+ expression.addValue(dataExtensionKey);
+ return expression;
+ }
+
+ private List fetchObjectData(ETClient client, Class clazz, ETFilter filter)
+ throws ETSdkException {
+
+ ClassLoader oldCL = Thread.currentThread().getContextClassLoader();
+ try {
+ Thread.currentThread().setContextClassLoader(MarketingCloudClient.class.getClassLoader());
+ //ETResponse etResponse = client.retrieve(clazz, new ETFilter());
+ ETResponse response = (ETResponse) PaginationETSoapObject.customRetrieve(client, null,
+ filter, null,
+ clazz);
+ List rows = response.getObjects();
+ while (response.getResponseMessage().equals("MoreDataAvailable")) {
+ response = (ETResponse) PaginationETSoapObject.customRetrieve(client, null,
+ filter, response.getRequestId(),
+ clazz);
+ rows.addAll(response.getObjects());
+ }
+ return rows;
+ } finally {
+ Thread.currentThread().setContextClassLoader(oldCL);
+ }
+ }
+
+
+ private List fetchObjectFields(Class clazz) {
+ return Arrays.stream(clazz.getDeclaredFields())
+ .map(o -> new MarketingCloudColumn(o.getName(), o.getType().getSimpleName().toUpperCase()))
+ .collect(Collectors.toList());
+ }
+
+ private T call(SFMCCall callable) throws ETSdkException {
+ ClassLoader oldClassloader = Thread.currentThread().getContextClassLoader();
+ try {
+ Thread.currentThread().setContextClassLoader(getClass().getClassLoader());
+ return callable.call();
+ } finally {
+ Thread.currentThread().setContextClassLoader(oldClassloader);
+ }
+ }
+
+ /**
+ * A SFMC call.
+ *
+ * @param type of return object
+ */
+ private interface SFMCCall {
+
+ /**
+ * Perform a call.
+ */
+ T call() throws ETSdkException;
+ }
+}
diff --git a/src/main/java/io/cdap/plugin/sfmc/source/MarketingCloudInputFormat.java b/src/main/java/io/cdap/plugin/sfmc/source/MarketingCloudInputFormat.java
new file mode 100644
index 0000000..a3d9eff
--- /dev/null
+++ b/src/main/java/io/cdap/plugin/sfmc/source/MarketingCloudInputFormat.java
@@ -0,0 +1,176 @@
+/*
+ * Copyright © 2020 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.sfmc.source;
+
+import io.cdap.cdap.api.data.format.StructuredRecord;
+import io.cdap.plugin.sfmc.source.util.MarketingCloudObjectInfo;
+import io.cdap.plugin.sfmc.source.util.SourceObject;
+import io.cdap.plugin.sfmc.source.util.SourceQueryMode;
+import io.cdap.plugin.sfmc.source.util.Util;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.JobContext;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Salesforce input format.
+ */
+public class MarketingCloudInputFormat extends InputFormat {
+ private static final Logger LOG = LoggerFactory.getLogger(MarketingCloudInputFormat.class);
+
+ /**
+ * Configure the input format to read tables from Salesforce. Should be called from the mapreduce client.
+ *
+ * @param jobConfig the job configuration
+ * @param mode the query mode
+ * @param conf the plugin conf
+ * @return Collection of MarketingCloudObjectInfo containing table and schema.
+ */
+ public static List setInput(Configuration jobConfig, SourceQueryMode mode,
+ MarketingCloudSourceConfig conf) {
+ MarketingCloudJobConfiguration jobConf = new MarketingCloudJobConfiguration(jobConfig);
+ jobConf.setPluginConfiguration(conf);
+
+ //Depending on the selected objects in the conf, get the schema for each object as MarketingCloudObjectInfo
+ List tableInfos = fetchTableInfo(mode, conf);
+
+ jobConf.setTableInfos(tableInfos);
+
+ LOG.debug("setInput::tableInfos = {}", tableInfos.size());
+ return tableInfos;
+ }
+
+ /**
+ * Depending on conf value fetch the list of fields for each object and create schema object.
+ *
+ * @param mode the query mode
+ * @param conf the plugin conf
+ * @return Collection of MarketingCloudObjectInfo containing table and schema.
+ */
+ private static List fetchTableInfo(SourceQueryMode mode, MarketingCloudSourceConfig conf) {
+ try {
+ MarketingCloudClient client = MarketingCloudClient.create(conf.getClientId(), conf.getClientSecret(),
+ conf.getAuthEndpoint(), conf.getSoapEndpoint());
+
+ boolean failOnError = conf.isFailOnError();
+
+ //When mode = SingleObject, fetch fields for the object selected in plugin config
+ if (mode == SourceQueryMode.SINGLE_OBJECT) {
+ MarketingCloudObjectInfo tableInfo = getTableMetaData(conf.getObject(), conf.getDataExtensionKey(), client,
+ failOnError);
+ return (tableInfo == null) ? Collections.emptyList() : Collections.singletonList(tableInfo);
+ }
+
+ //When mode = MultiObject, get the list of objects provided in plugin config and the fetch fields for each of
+ //then objects.
+ List tableInfos = new ArrayList<>();
+ List objectList = conf.getObjectList();
+
+ for (SourceObject object : objectList) {
+ MarketingCloudObjectInfo tableInfo = null;
+
+ if (object == SourceObject.DATA_EXTENSION) {
+ //if the object = Data Extension then get the list of data extension keys and then fetch fields for each of
+ //the data extension objects.
+ List dataExtensionKeys = Util.splitToList(conf.getDataExtensionKeys(), ',');
+ for (String dataExtensionKey : dataExtensionKeys) {
+ tableInfo = getTableMetaData(object, dataExtensionKey, client, failOnError);
+ if (tableInfo == null) {
+ continue;
+ }
+ tableInfos.add(tableInfo);
+ }
+ } else {
+ tableInfo = getTableMetaData(object, "", client, failOnError);
+ if (tableInfo != null) {
+ tableInfos.add(tableInfo);
+ }
+ }
+ }
+
+ return tableInfos;
+ } catch (Exception e) {
+ if (conf.isFailOnError()) {
+ LOG.error("Error in fetchTableInfo()", e);
+ } else {
+ LOG.warn("Failed in fetchTableInfo()", e);
+ }
+ return Collections.emptyList();
+ }
+ }
+
+ /**
+ * Fetch the fields for passed object.
+ */
+ private static MarketingCloudObjectInfo getTableMetaData(SourceObject object, String dataExtensionKey,
+ MarketingCloudClient client, boolean failOnError) {
+ try {
+ if (object == SourceObject.DATA_EXTENSION) {
+ return client.fetchDataExtensionSchema(dataExtensionKey);
+ } else {
+ return client.fetchObjectSchema(object);
+ }
+ } catch (Exception e) {
+ if (failOnError) {
+ LOG.error("Error in getTableMetaData()", e);
+ } else {
+ LOG.warn("Failed in getTableMetaData()", e);
+ }
+ return null;
+ }
+ }
+
+ @Override
+ public List getSplits(JobContext jobContext) throws IOException, InterruptedException {
+ MarketingCloudJobConfiguration jobConfig = new MarketingCloudJobConfiguration(jobContext.getConfiguration());
+ MarketingCloudSourceConfig pluginConf = jobConfig.getPluginConf();
+
+ List tableInfos = jobConfig.getTableInfos();
+ List resultSplits = new ArrayList<>();
+
+ for (MarketingCloudObjectInfo tableInfo : tableInfos) {
+ String tableKey = tableInfo.getObject().name();
+ String tableName = tableInfo.getTableName();
+ resultSplits.add(new MarketingCloudInputSplit(tableKey, tableName));
+ }
+
+ LOG.debug("# of split = {}", resultSplits.size());
+
+ return resultSplits;
+ }
+
+ @Override
+ public RecordReader createRecordReader(InputSplit inputSplit,
+ TaskAttemptContext taskAttemptContext)
+ throws IOException, InterruptedException {
+ MarketingCloudJobConfiguration jobConfig = new MarketingCloudJobConfiguration(
+ taskAttemptContext.getConfiguration());
+ MarketingCloudSourceConfig pluginConf = jobConfig.getPluginConf();
+
+ return new MarketingCloudRecordReader(pluginConf);
+ }
+}
diff --git a/src/main/java/io/cdap/plugin/sfmc/source/MarketingCloudInputSplit.java b/src/main/java/io/cdap/plugin/sfmc/source/MarketingCloudInputSplit.java
new file mode 100644
index 0000000..396409e
--- /dev/null
+++ b/src/main/java/io/cdap/plugin/sfmc/source/MarketingCloudInputSplit.java
@@ -0,0 +1,78 @@
+/*
+ * Copyright © 2020 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.sfmc.source;
+
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.mapreduce.InputSplit;
+
+import java.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+/**
+ * Salesforce input split.
+ */
+public class MarketingCloudInputSplit extends InputSplit implements Writable {
+ private String objectName;
+ private String tableName;
+
+ // used by mapreduce
+ public MarketingCloudInputSplit() {
+ }
+
+ /**
+ * Constructor for MarketingCloudInputSplit.
+ *
+ * @param objectName The object name
+ * @param tableName The corresponding table name
+ */
+ public MarketingCloudInputSplit(String objectName, String tableName) {
+ this.objectName = objectName;
+ this.tableName = tableName;
+ }
+
+ public String getObjectName() {
+ return objectName;
+ }
+
+ public String getTableName() {
+ return tableName;
+ }
+
+
+ @Override
+ public void write(DataOutput dataOutput) throws IOException {
+ dataOutput.writeUTF(this.objectName);
+ dataOutput.writeUTF(this.tableName);
+ }
+
+ @Override
+ public void readFields(DataInput dataInput) throws IOException {
+ this.objectName = dataInput.readUTF();
+ this.tableName = dataInput.readUTF();
+ }
+
+ @Override
+ public long getLength() throws IOException, InterruptedException {
+ return 0;
+ }
+
+ @Override
+ public String[] getLocations() throws IOException, InterruptedException {
+ return new String[0];
+ }
+}
diff --git a/src/main/java/io/cdap/plugin/sfmc/source/MarketingCloudJobConfiguration.java b/src/main/java/io/cdap/plugin/sfmc/source/MarketingCloudJobConfiguration.java
new file mode 100644
index 0000000..42d42cf
--- /dev/null
+++ b/src/main/java/io/cdap/plugin/sfmc/source/MarketingCloudJobConfiguration.java
@@ -0,0 +1,70 @@
+/*
+ * Copyright © 2020 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.sfmc.source;
+
+import com.google.gson.Gson;
+import com.google.gson.reflect.TypeToken;
+import io.cdap.plugin.sfmc.source.util.MarketingCloudObjectInfo;
+import org.apache.hadoop.conf.Configuration;
+
+import java.lang.reflect.Type;
+import java.util.List;
+
+/**
+ * Allows to specify and access connection configuration properties of {@link Configuration}.
+ */
+public class MarketingCloudJobConfiguration {
+ private static final String PLUGIN_CONF_FIELD = "salesforce.plugin.conf";
+ private static final String INFO_FIELD = "salesforce.table.info";
+
+ private static final Type PLUGIN_CONF_TYPE = new TypeToken() {
+ }.getType();
+ private static final Type INFO_TYPE = new TypeToken>() {
+ }.getType();
+
+ private static final Gson GSON = new Gson();
+
+ private Configuration conf;
+
+ public MarketingCloudJobConfiguration(Configuration job) {
+ this.conf = job;
+ }
+
+ public Configuration getConf() {
+ return this.conf;
+ }
+
+ public void setPluginConfiguration(MarketingCloudSourceConfig conf) {
+ set(PLUGIN_CONF_FIELD, GSON.toJson(conf));
+ }
+
+ public MarketingCloudSourceConfig getPluginConf() {
+ return GSON.fromJson(getConf().get(PLUGIN_CONF_FIELD), PLUGIN_CONF_TYPE);
+ }
+
+ public List getTableInfos() {
+ return GSON.fromJson(getConf().get(INFO_FIELD), INFO_TYPE);
+ }
+
+ public void setTableInfos(List infoList) {
+ set(INFO_FIELD, GSON.toJson(infoList));
+ }
+
+ private void set(String key, String value) {
+ getConf().set(key, value);
+ }
+}
diff --git a/src/main/java/io/cdap/plugin/sfmc/source/MarketingCloudRecordReader.java b/src/main/java/io/cdap/plugin/sfmc/source/MarketingCloudRecordReader.java
new file mode 100644
index 0000000..2b73105
--- /dev/null
+++ b/src/main/java/io/cdap/plugin/sfmc/source/MarketingCloudRecordReader.java
@@ -0,0 +1,319 @@
+/*
+ * Copyright © 2020 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.sfmc.source;
+
+import com.exacttarget.fuelsdk.ETApiObject;
+import com.exacttarget.fuelsdk.ETDataExtensionRow;
+import com.google.common.base.Strings;
+import io.cdap.cdap.api.data.format.StructuredRecord;
+import io.cdap.cdap.api.data.schema.Schema;
+import io.cdap.plugin.sfmc.source.util.MarketingCloudConstants;
+import io.cdap.plugin.sfmc.source.util.MarketingCloudObjectInfo;
+import io.cdap.plugin.sfmc.source.util.SourceObject;
+import io.cdap.plugin.sfmc.source.util.SourceQueryMode;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.mapreduce.InputSplit;
+import org.apache.hadoop.mapreduce.RecordReader;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Date;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Record reader that reads the entire contents of a Salesforce table.
+ */
+public class MarketingCloudRecordReader extends RecordReader {
+ private static final Logger LOG = LoggerFactory.getLogger(MarketingCloudRecordReader.class);
+ private final MarketingCloudSourceConfig pluginConf;
+ private MarketingCloudInputSplit split;
+ private int pos;
+ private List tableFields;
+ private MarketingCloudObjectInfo sfObjectMetaData;
+ private Schema schema;
+
+ private SourceObject object;
+ private String dataExtensionKey = "";
+ private String tableName;
+ private String formattedTableName;
+ private String tableNameField;
+ private List results;
+ private Iterator iterator;
+ private ETApiObject row;
+
+ MarketingCloudRecordReader(MarketingCloudSourceConfig pluginConf) {
+ this.pluginConf = pluginConf;
+ }
+
+ @Override
+ public void initialize(InputSplit split, TaskAttemptContext context) {
+ this.split = (MarketingCloudInputSplit) split;
+ this.pos = 0;
+ }
+
+ @Override
+ public boolean nextKeyValue() throws IOException {
+ try {
+ if (results == null) {
+ fetchData();
+ }
+
+ if (!iterator.hasNext()) {
+ return false;
+ }
+
+ row = iterator.next();
+
+ pos++;
+ } catch (Exception e) {
+ if (pluginConf.isFailOnError()) {
+ LOG.error("Error in nextKeyValue", e);
+ throw new IOException("Exception in nextKeyValue", e);
+ } else {
+ LOG.warn("Failed in nextKeyValue", e);
+ }
+ }
+ return true;
+ }
+
+ @Override
+ public NullWritable getCurrentKey() {
+ return NullWritable.get();
+ }
+
+ @Override
+ public StructuredRecord getCurrentValue() throws IOException {
+ StructuredRecord.Builder recordBuilder = StructuredRecord.builder(schema);
+
+ if (pluginConf.getQueryMode() == SourceQueryMode.MULTI_OBJECT) {
+ recordBuilder.set(tableNameField, formattedTableName);
+ }
+
+ try {
+
+ convertRecord(recordBuilder, row);
+
+ } catch (Exception e) {
+ if (pluginConf.isFailOnError()) {
+ LOG.error(String.format("Error decoding row from table %s", tableName), e);
+ throw new IOException(String.format("Error decoding row from table %s", tableName), e);
+ } else {
+ LOG.warn(String.format("Failed decoding row from table %s", tableName), e);
+ }
+ }
+ return recordBuilder.build();
+ }
+
+ @Override
+ public float getProgress() throws IOException, InterruptedException {
+ return pos / (float) split.getLength();
+ }
+
+ @Override
+ public void close() throws IOException {
+ }
+
+ private void fetchData() {
+ object = SourceObject.valueOf(split.getObjectName());
+ tableName = split.getTableName();
+ formattedTableName = tableName.replaceAll("-", "_");
+
+ if (object == SourceObject.DATA_EXTENSION) {
+ dataExtensionKey = tableName.replaceAll(MarketingCloudConstants.DATA_EXTENSION_PREFIX, "");
+ }
+ tableNameField = pluginConf.getTableNameField();
+
+ //Get the table data
+ try {
+ MarketingCloudClient client = MarketingCloudClient.create(pluginConf.getClientId(), pluginConf.getClientSecret(),
+ pluginConf.getAuthEndpoint(),
+ pluginConf.getSoapEndpoint());
+
+ //Fetch data
+ if (object == SourceObject.DATA_EXTENSION) {
+ results = client.fetchDataExtensionRecords(dataExtensionKey, object.getFilter());
+ } else {
+ results = client.fetchObjectRecords(object);
+ }
+
+ LOG.debug("size={}", results.size());
+ if (!results.isEmpty()) {
+ fetchSchema(client);
+ }
+ } catch (Exception e) {
+ results = Collections.emptyList();
+ if (pluginConf.isFailOnError()) {
+ LOG.error("Error while fetching data", e);
+ } else {
+ LOG.warn("Failed while fetching data", e);
+ }
+ }
+
+ iterator = results.iterator();
+ }
+
+ private void fetchSchema(MarketingCloudClient client) {
+ //Fetch the column definition
+ List schemaFields;
+
+ try {
+ if (object == SourceObject.DATA_EXTENSION) {
+ sfObjectMetaData = client.fetchDataExtensionSchema(dataExtensionKey);
+ } else {
+ sfObjectMetaData = client.fetchObjectSchema(object);
+ }
+
+ //Build schema
+ tableFields = sfObjectMetaData.getSchema().getFields();
+ schemaFields = new ArrayList<>(tableFields);
+
+ if (pluginConf.getQueryMode() == SourceQueryMode.MULTI_OBJECT) {
+ schemaFields.add(Schema.Field.of(tableNameField, Schema.of(Schema.Type.STRING)));
+ }
+ } catch (Exception e) {
+ schemaFields = Collections.emptyList();
+ }
+
+ schema = Schema.recordOf(tableName.replaceAll("-", "_"), schemaFields);
+ }
+
+ /**
+ * Read data from ETApiObject and convert it to StructureRecord.
+ */
+ private void convertRecord(StructuredRecord.Builder recordBuilder, ETApiObject row) {
+ for (Schema.Field field : tableFields) {
+ String fieldName = field.getName();
+ Object rawFieldValue = null;
+ if (row instanceof ETDataExtensionRow) {
+ String apiFieldName = sfObjectMetaData.lookupFieldsMap(fieldName);
+ rawFieldValue = ((ETDataExtensionRow) row).getColumn(apiFieldName);
+ } else {
+ rawFieldValue = getFieldValue(row, fieldName);
+ }
+
+ Object fieldValue = convertToValue(fieldName, field.getSchema(), rawFieldValue);
+
+ recordBuilder.set(fieldName, fieldValue);
+ }
+ }
+
+ /**
+ * Read data from ETApiObject using reflection for a given field name.
+ */
+ private Object getFieldValue(ETApiObject row, String fieldName) {
+ try {
+ Method method = row.getClass().getMethod(createGetterName(fieldName));
+ return method.invoke(row);
+ } catch (Exception e) {
+ if (pluginConf.isFailOnError()) {
+ LOG.error(String.format("Error while fetching %s.%s value", row.getClass().getSimpleName(), fieldName), e);
+ } else {
+ LOG.warn(String.format("Failed while fetching %s.%s value", row.getClass().getSimpleName(), fieldName), e);
+ }
+ return null;
+ }
+ }
+
+ /**
+ * Constructs the get method name to be used in reflection call.
+ */
+ private String createGetterName(String name) {
+ StringBuilder sb = new StringBuilder("get");
+ sb.append(name.substring(0, 1).toUpperCase());
+ sb.append(name.substring(1));
+ return sb.toString();
+ }
+
+ /**
+ * Converts raw field value according to the schema field type.
+ */
+ private Object convertToValue(String fieldName, Schema fieldSchema, Object fieldValue) {
+ Schema.Type fieldType = fieldSchema.getType();
+ Schema.LogicalType logicalType = fieldSchema.getLogicalType();
+ if (fieldSchema.getLogicalType() != null) {
+ return transformLogicalType(fieldName, logicalType, fieldValue);
+ }
+
+ switch (fieldType) {
+ case STRING:
+ return convertToStringValue(fieldValue);
+ case DOUBLE:
+ return convertToDoubleValue(fieldValue);
+ case INT:
+ return convertToIntegerValue(fieldValue);
+ case BOOLEAN:
+ return convertToBooleanValue(fieldValue);
+ case UNION:
+ if (fieldSchema.isNullable()) {
+ return convertToValue(fieldName, fieldSchema.getNonNullable(), fieldValue);
+ }
+ throw new IllegalStateException(
+ String.format("Field '%s' is of unexpected type '%s'. Declared 'complex UNION' types: %s",
+ fieldName, fieldValue.getClass().getSimpleName(), fieldSchema.getUnionSchemas()));
+ default:
+ throw new IllegalStateException(
+ String.format("Record type '%s' is not supported for field '%s'", fieldType.name(), fieldName));
+ }
+ }
+
+ private String convertToStringValue(Object fieldValue) {
+ return String.valueOf(fieldValue);
+ }
+
+ private Double convertToDoubleValue(Object fieldValue) {
+ if (fieldValue instanceof String && Strings.isNullOrEmpty(String.valueOf(fieldValue))) {
+ return null;
+ }
+
+ return Double.parseDouble(String.valueOf(fieldValue));
+ }
+
+ private Integer convertToIntegerValue(Object fieldValue) {
+ if (fieldValue instanceof String && Strings.isNullOrEmpty(String.valueOf(fieldValue))) {
+ return null;
+ }
+
+ return Integer.parseInt(String.valueOf(fieldValue));
+ }
+
+ private Boolean convertToBooleanValue(Object fieldValue) {
+ if (fieldValue instanceof String && Strings.isNullOrEmpty(String.valueOf(fieldValue))) {
+ return null;
+ }
+
+ return Boolean.parseBoolean(String.valueOf(fieldValue));
+ }
+ private Object transformLogicalType(String fieldName, Schema.LogicalType logicalType, Object value) {
+ switch (logicalType) {
+ case TIMESTAMP_MICROS:
+ if (value instanceof Date) {
+ return TimeUnit.MILLISECONDS.toMicros((((Date) value).getTime()));
+ }
+ return null;
+ default:
+ throw new IllegalArgumentException(
+ String.format("Field '%s' is of unsupported type '%s'", fieldName, logicalType.getToken()));
+ }
+ }
+}
diff --git a/src/main/java/io/cdap/plugin/sfmc/source/MarketingCloudSource.java b/src/main/java/io/cdap/plugin/sfmc/source/MarketingCloudSource.java
new file mode 100644
index 0000000..85f2352
--- /dev/null
+++ b/src/main/java/io/cdap/plugin/sfmc/source/MarketingCloudSource.java
@@ -0,0 +1,148 @@
+/*
+ * Copyright © 2020 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.sfmc.source;
+
+import io.cdap.cdap.api.annotation.Description;
+import io.cdap.cdap.api.annotation.Name;
+import io.cdap.cdap.api.annotation.Plugin;
+import io.cdap.cdap.api.data.batch.Input;
+import io.cdap.cdap.api.data.format.StructuredRecord;
+import io.cdap.cdap.api.data.schema.Schema;
+import io.cdap.cdap.api.dataset.lib.KeyValue;
+import io.cdap.cdap.etl.api.Emitter;
+import io.cdap.cdap.etl.api.Engine;
+import io.cdap.cdap.etl.api.FailureCollector;
+import io.cdap.cdap.etl.api.PipelineConfigurer;
+import io.cdap.cdap.etl.api.StageConfigurer;
+import io.cdap.cdap.etl.api.action.SettableArguments;
+import io.cdap.cdap.etl.api.batch.BatchSource;
+import io.cdap.cdap.etl.api.batch.BatchSourceContext;
+import io.cdap.plugin.common.LineageRecorder;
+import io.cdap.plugin.common.SourceInputFormatProvider;
+import io.cdap.plugin.sfmc.source.util.MarketingCloudConstants;
+import io.cdap.plugin.sfmc.source.util.MarketingCloudObjectInfo;
+import io.cdap.plugin.sfmc.source.util.SourceQueryMode;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.NullWritable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+/**
+ * A {@link BatchSource} that reads data from multiple objects in Salesforce.
+ */
+@Plugin(type = BatchSource.PLUGIN_TYPE)
+@Name(MarketingCloudConstants.PLUGIN_NAME)
+@Description("Reads from multiple objects in Salesforce. " +
+ "Outputs one record for each row in each object, with the object name as a record field. " +
+ "Also sets a pipeline argument for each object read, which contains the object schema.")
+public class MarketingCloudSource extends BatchSource {
+ private static final Logger LOG = LoggerFactory.getLogger(MarketingCloudSource.class);
+
+ private final MarketingCloudSourceConfig conf;
+
+ public MarketingCloudSource(MarketingCloudSourceConfig conf) {
+ this.conf = conf;
+ }
+
+ @Override
+ public void configurePipeline(PipelineConfigurer pipelineConfigurer) {
+ super.configurePipeline(pipelineConfigurer);
+
+ StageConfigurer stageConfigurer = pipelineConfigurer.getStageConfigurer();
+ FailureCollector collector = stageConfigurer.getFailureCollector();
+
+ conf.validate(collector);
+ // Since we have validated all the properties, throw an exception if there are any
+ // errors in the collector. This is to avoid adding same validation errors again in
+ // getSchema method call
+ collector.getOrThrowException();
+
+ //Get Schema
+ stageConfigurer.setOutputSchema(getSchema(conf.getQueryMode()));
+
+ if (conf.isFailOnError()) {
+ if (pipelineConfigurer.getEngine() == Engine.SPARK) {
+ pipelineConfigurer.setPipelineProperties(Collections.singletonMap("spark.task.maxFailures", "1"));
+ } else if (pipelineConfigurer.getEngine() == Engine.MAPREDUCE) {
+ Map properties = new HashMap<>();
+ properties.put("mapreduce.reduce.maxattempts", "1");
+ properties.put("mapreduce.map.maxattempts", "1");
+ pipelineConfigurer.setPipelineProperties(properties);
+ }
+ }
+ }
+
+ @Override
+ public void prepareRun(BatchSourceContext context) throws Exception {
+ FailureCollector collector = context.getFailureCollector();
+ conf.validate(collector);
+ collector.getOrThrowException();
+
+ SourceQueryMode mode = conf.getQueryMode(collector);
+
+ Configuration hConf = new Configuration();
+ Collection tables = MarketingCloudInputFormat.setInput(hConf, mode, conf);
+ SettableArguments arguments = context.getArguments();
+ for (MarketingCloudObjectInfo tableInfo : tables) {
+ arguments.set(MarketingCloudConstants.TABLE_PREFIX + tableInfo.getFormattedTableName(),
+ tableInfo.getSchema().toString());
+ recordLineage(context, tableInfo);
+ }
+
+ context.setInput(Input.of(conf.getReferenceName(),
+ new SourceInputFormatProvider(MarketingCloudInputFormat.class, hConf)));
+ }
+
+ @Override
+ public void transform(KeyValue input, Emitter emitter) {
+ emitter.emit(input.getValue());
+ }
+
+ private Schema getSchema(SourceQueryMode mode) {
+ Schema schema = null;
+ if (mode == SourceQueryMode.SINGLE_OBJECT) {
+ Configuration hConf = new Configuration();
+ Collection tables = MarketingCloudInputFormat.setInput(hConf, mode, conf);
+ if (tables != null && !tables.isEmpty()) {
+ schema = tables.iterator().next().getSchema();
+ }
+ }
+ return schema;
+ }
+
+ private void recordLineage(BatchSourceContext context, MarketingCloudObjectInfo tableInfo) {
+ String tableName = tableInfo.getFormattedTableName();
+ String outputName = String.format("%s-%s", conf.getReferenceName(), tableName);
+ Schema schema = tableInfo.getSchema();
+ LineageRecorder lineageRecorder = new LineageRecorder(context, outputName);
+ lineageRecorder.createExternalDataset(schema);
+ List fields = Objects.requireNonNull(schema).getFields();
+ if (fields != null && !fields.isEmpty()) {
+ lineageRecorder.recordRead("Read",
+ String.format("Read from '%s' Marketing Cloud object.", tableName),
+ fields.stream().map(Schema.Field::getName).collect(Collectors.toList()));
+ }
+ }
+}
diff --git a/src/main/java/io/cdap/plugin/sfmc/source/MarketingCloudSourceConfig.java b/src/main/java/io/cdap/plugin/sfmc/source/MarketingCloudSourceConfig.java
new file mode 100644
index 0000000..80c4a72
--- /dev/null
+++ b/src/main/java/io/cdap/plugin/sfmc/source/MarketingCloudSourceConfig.java
@@ -0,0 +1,487 @@
+/*
+ * Copyright © 2020 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.sfmc.source;
+
+import com.exacttarget.fuelsdk.ETSdkException;
+import com.google.common.annotations.VisibleForTesting;
+import io.cdap.cdap.api.annotation.Description;
+import io.cdap.cdap.api.annotation.Macro;
+import io.cdap.cdap.api.annotation.Name;
+import io.cdap.cdap.api.plugin.PluginConfig;
+import io.cdap.cdap.etl.api.FailureCollector;
+import io.cdap.plugin.common.Constants;
+import io.cdap.plugin.common.IdUtils;
+import io.cdap.plugin.sfmc.source.util.MarketingCloudConstants;
+import io.cdap.plugin.sfmc.source.util.SourceObject;
+import io.cdap.plugin.sfmc.source.util.SourceQueryMode;
+import io.cdap.plugin.sfmc.source.util.Util;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import javax.annotation.Nullable;
+
+/**
+ * Configuration for the {@link MarketingCloudSource}.
+ */
+public class MarketingCloudSourceConfig extends PluginConfig {
+ private static final Logger LOG = LoggerFactory.getLogger(MarketingCloudSourceConfig.class);
+
+ @Name(Constants.Reference.REFERENCE_NAME)
+ @Description("This will be used to uniquely identify this source for lineage, annotating metadata, etc.")
+ private String referenceName;
+
+ @Name(MarketingCloudConstants.PROPERTY_QUERY_MODE)
+ @Macro
+ @Description("Mode of data retrieval. The mode can be one of two values: "
+ + "`Multi Object` - will allow user to fetch data for multiple data extensions, "
+ + "`Single Object` - will allow user to fetch data for single data extension.")
+ private String queryMode;
+
+ @Name(MarketingCloudConstants.PROPERTY_OBJECT_NAME)
+ @Macro
+ @Nullable
+ @Description("Specify the object for which data to be fetched. This can be one of following values: " +
+ "`Data Extension` - will allow user to fetch data for a single Data Extension object, " +
+ "`Campaign` - will allow user to fetch data for Campaign object, " +
+ "`Email` - will allow user to fetch data for Email object, " +
+ "`Mailing List` - will allow user to fetch data for Mailing List object. " +
+ "Note, this value will be ignored if the Mode is set to `Multi Object`.")
+ private String objectName;
+
+ @Name(MarketingCloudConstants.PROPERTY_DATA_EXTENSION_KEY)
+ @Macro
+ @Nullable
+ @Description("Specify the data extension key from which data to be fetched. Note, this value will be ignored in " +
+ "following two cases: 1. If the Mode is set to `Multi Object`, 2. If the selected object name is other than " +
+ "`Data Extension`.")
+ private String dataExtensionKey;
+
+ @Name(MarketingCloudConstants.PROPERTY_OBJECT_LIST)
+ @Macro
+ @Nullable
+ @Description("Specify the comma-separated list of objects for which data to be fetched; for example: " +
+ "'Object1,Object2'. This can be one or more values from following possible values: " +
+ "`Data Extension` - will allow user to fetch data for a single Data Extension object, " +
+ "`Campaign` - will allow user to fetch data for Campaign object, " +
+ "`Email` - will allow user to fetch data for Email object, " +
+ "`Mailing List` - will allow user to fetch data for Mailing List object. " +
+ "Note, this value will be ignored if the Mode is set to `Single Object`.")
+ private String objectList;
+
+ @Name(MarketingCloudConstants.PROPERTY_DATA_EXTENSION_KEY_LIST)
+ @Macro
+ @Nullable
+ @Description("Specify the data extension keys from which data to be fetched; for example: 'Key1,Key2'. " +
+ "Note, this value will be ignored in following two cases: 1. If the Mode is set to `Single Object`, " +
+ "2. If the selected object list does not contain `Data Extension` as one of the objects.")
+ private String dataExtensionKeys;
+
+ @Name(MarketingCloudConstants.PROPERTY_TABLE_NAME_FIELD)
+ @Macro
+ @Nullable
+ @Description("The name of the field that holds the object name to which the data belongs to. Must not be the name " +
+ "of any column for any of the objects that will be read. Defaults to `tablename`. In case of `Data Extension` " +
+ "object, this field will have value in `dataextension_[Data Extension Key]` format. Note, the Table name field " +
+ "value will be ignored if the Mode is set to `Single Object`.")
+ private String tableNameField;
+
+ @Name(MarketingCloudConstants.PROPERTY_FILTER)
+ @Macro
+ @Nullable
+ @Description("The WHERE clause used to filter data from Marketing cloud objects.")
+ private String filter;
+
+ @Name(MarketingCloudConstants.PROPERTY_CLIENT_ID)
+ @Macro
+ @Description("OAuth2 client ID associated with an installed package in the Salesforce Marketing Cloud.")
+ private String clientId;
+
+ @Name(MarketingCloudConstants.PROPERTY_CLIENT_SECRET)
+ @Macro
+ @Description("OAuth2 client secret associated with an installed package in the Salesforce Marketing Cloud.")
+ private String clientSecret;
+
+ @Name(MarketingCloudConstants.PROPERTY_API_ENDPOINT)
+ @Macro
+ @Description("The REST API Base URL associated for the Server-to-Server API integration. " +
+ "For example, https://instance.rest.marketingcloudapis.com/")
+ private String restEndpoint;
+
+ @Name(MarketingCloudConstants.PROPERTY_AUTH_API_ENDPOINT)
+ @Macro
+ @Description("Authentication Base URL associated for the Server-to-Server API integration. " +
+ "For example, https://instance.auth.marketingcloudapis.com/")
+ private String authEndpoint;
+
+ @Name(MarketingCloudConstants.PROPERTY_SOAP_API_ENDPOINT)
+ @Macro
+ @Description("The SOAP Endpoint URL associated for the Server-to-Server API integration. " +
+ "For example, https://instance.soap.marketingcloudapis.com/Service.asmx")
+ private String soapEndpoint;
+
+ @Macro
+ @Nullable
+ @Name(MarketingCloudConstants.PROPERTY_FAIL_ON_ERROR)
+ @Description("Whether to fail the pipeline if it fails while reading.")
+ private Boolean failOnError;
+
+ /**
+ * Constructor for MarketingCloudSourceConfig object.
+ *
+ * @param referenceName The reference name
+ * @param queryMode The query mode
+ * @param objectName The object name to be fetched from Salesforce Marketing Cloud
+ * @param dataExtensionKey The data extension key to be fetched from Salesforce Marketing Cloud
+ * @param objectList The list of objects to be fetched from Salesforce Marketing Cloud
+ * @param dataExtensionKeys The list of data extension keys to be fetched from Salesforce Marketing Cloud
+ * @param tableNameField The field name to hold the table name value
+ * @param clientId The Salesforce Marketing Cloud Client Id
+ * @param clientSecret The Salesforce Marketing Cloud Client Secret
+ * @param restEndpoint The REST API endpoint for Salesforce Marketing Cloud
+ * @param authEndpoint The AUTH API endpoint for Salesforce Marketing Cloud
+ * @param soapEndpoint The SOAP API endpoint for Salesforce Marketing Cloud
+ * @param failOnError A flag to decide what is to be done to the pipeline execution in case of errors
+ */
+ public MarketingCloudSourceConfig(String referenceName, String queryMode, @Nullable String objectName,
+ @Nullable String dataExtensionKey, @Nullable String objectList,
+ @Nullable String dataExtensionKeys, @Nullable String tableNameField,
+ @Nullable String filter, String clientId, String clientSecret,
+ String restEndpoint, String authEndpoint, String soapEndpoint,
+ Boolean failOnError) {
+ this.referenceName = referenceName;
+ this.queryMode = queryMode;
+ this.objectName = objectName;
+ this.dataExtensionKey = dataExtensionKey;
+ this.objectList = objectList;
+ this.dataExtensionKeys = dataExtensionKeys;
+ this.tableNameField = tableNameField;
+ this.filter = filter;
+ this.clientId = clientId;
+ this.clientSecret = clientSecret;
+ this.restEndpoint = restEndpoint;
+ this.authEndpoint = authEndpoint;
+ this.soapEndpoint = soapEndpoint;
+ this.failOnError = failOnError;
+ }
+
+ public String getReferenceName() {
+ return referenceName;
+ }
+
+ /**
+ * Returns the query mode chosen.
+ *
+ * @param collector The failure collector to collect the errors
+ * @return An instance of SourceQueryMode
+ */
+ public SourceQueryMode getQueryMode(FailureCollector collector) {
+ SourceQueryMode mode = getQueryMode();
+ if (mode != null) {
+ return mode;
+ }
+
+ collector.addFailure("Unsupported query mode value: " + queryMode,
+ String.format("Supported modes are: %s", SourceQueryMode.getSupportedModes()))
+ .withConfigProperty(MarketingCloudConstants.PROPERTY_QUERY_MODE);
+ collector.getOrThrowException();
+ return null;
+ }
+
+ /**
+ * Returns the query mode chosen.
+ *
+ * @return An instance of SourceQueryMode
+ */
+ public SourceQueryMode getQueryMode() {
+ Optional sourceQueryMode = SourceQueryMode.fromValue(queryMode);
+
+ return sourceQueryMode.isPresent() ? sourceQueryMode.get() : null;
+ }
+
+ /**
+ * Returns selected object.
+ *
+ * @param collector The failure collector to collect the errors
+ * @return An instance of SourceObject
+ */
+ public SourceObject getObject(FailureCollector collector) {
+ SourceObject sourceObject = getObject();
+ if (sourceObject != null) {
+ return sourceObject;
+ }
+
+ collector.addFailure("Unsupported object value: " + objectName,
+ String.format("Supported objects are: %s", SourceObject.getSupportedObjects()))
+ .withConfigProperty(MarketingCloudConstants.PROPERTY_OBJECT_NAME);
+ collector.getOrThrowException();
+ return null;
+ }
+
+ @Nullable
+ public SourceObject getObject() {
+ return getSourceObject(objectName, filter);
+ }
+
+ @Nullable
+ public String getDataExtensionKey() {
+ return dataExtensionKey;
+ }
+
+ /**
+ * Returns list of selected objects.
+ *
+ * @param collector The failure collector to collect the errors
+ * @return The list of SourceObject
+ */
+ public List getObjectList(FailureCollector collector) {
+ List objects = Util.splitToList(objectList, ',');
+ List sourceObjects = new ArrayList<>();
+
+ for (String object : objects) {
+ SourceObject sourceObject = getSourceObject(object, filter);
+ if (sourceObject == null) {
+ collector.addFailure("Unsupported object value: " + object,
+ String.format("Supported objects are: %s", SourceObject.getSupportedObjects()))
+ .withConfigProperty(MarketingCloudConstants.PROPERTY_OBJECT_LIST);
+ break;
+ }
+ sourceObjects.add(sourceObject);
+ }
+
+ return sourceObjects;
+ }
+
+ /**
+ * Returns list of selected objects.
+ *
+ * @return The list of SourceObject
+ */
+ @Nullable
+ public List getObjectList() {
+ List objects = Util.splitToList(objectList, ',');
+ List sourceObjects = new ArrayList<>();
+
+ for (String object : objects) {
+ SourceObject sourceObject = getSourceObject(object, filter);
+ if (sourceObject == null) {
+ continue;
+ }
+ sourceObjects.add(sourceObject);
+ }
+
+ return sourceObjects;
+ }
+
+ @Nullable
+ public String getDataExtensionKeys() {
+ return dataExtensionKeys;
+ }
+
+ @Nullable
+ public String getTableNameField() {
+ return tableNameField;
+ }
+
+ @Nullable
+ public String getFilter() {
+ return filter;
+ }
+
+ public String getClientId() {
+ return clientId;
+ }
+
+ public String getClientSecret() {
+ return clientSecret;
+ }
+
+ public String getRestEndpoint() {
+ return restEndpoint;
+ }
+
+ public String getAuthEndpoint() {
+ return authEndpoint;
+ }
+
+ public String getSoapEndpoint() {
+ return soapEndpoint;
+ }
+
+ public Boolean isFailOnError() {
+ return failOnError == null ? false : failOnError;
+ }
+
+ /**
+ * Validates {@link MarketingCloudSourceConfig} instance.
+ */
+ public void validate(FailureCollector collector) {
+ //Validates the given referenceName to consists of characters allowed to represent a dataset.
+ IdUtils.validateReferenceName(referenceName, collector);
+
+ validateCredentials(collector);
+ validateQueryMode(collector);
+ validateFilter(collector);
+ }
+
+ private SourceObject getSourceObject(String objectName, String filter) {
+ Optional sourceObject = SourceObject.fromValue(objectName);
+ if (sourceObject.isPresent()) {
+ SourceObject obj = sourceObject.get();
+ obj.setFilter(filter);
+ return obj;
+ } else {
+ return null;
+ }
+ }
+
+ private void validateCredentials(FailureCollector collector) {
+ if (!shouldConnect()) {
+ return;
+ }
+
+ if (Util.isNullOrEmpty(clientId)) {
+ collector.addFailure("Client ID must be specified.", null)
+ .withConfigProperty(MarketingCloudConstants.PROPERTY_CLIENT_ID);
+ }
+
+ if (Util.isNullOrEmpty(clientSecret)) {
+ collector.addFailure("Client Secret must be specified.", null)
+ .withConfigProperty(MarketingCloudConstants.PROPERTY_CLIENT_SECRET);
+ }
+
+ if (Util.isNullOrEmpty(restEndpoint)) {
+ collector.addFailure(" REST Endpoint must be specified.", null)
+ .withConfigProperty(MarketingCloudConstants.PROPERTY_API_ENDPOINT);
+ }
+
+ if (Util.isNullOrEmpty(authEndpoint)) {
+ collector.addFailure("Auth Endpoint must be specified.", null)
+ .withConfigProperty(MarketingCloudConstants.PROPERTY_AUTH_API_ENDPOINT);
+ }
+
+ if (Util.isNullOrEmpty(soapEndpoint)) {
+ collector.addFailure("Soap Endpoint must be specified.", null)
+ .withConfigProperty(MarketingCloudConstants.PROPERTY_SOAP_API_ENDPOINT);
+ }
+
+ collector.getOrThrowException();
+ validateSalesforceConnection(collector);
+ }
+
+ @VisibleForTesting
+ void validateSalesforceConnection(FailureCollector collector) {
+ try {
+ MarketingCloudClient.create(clientId, clientSecret, authEndpoint, soapEndpoint);
+ } catch (ETSdkException e) {
+ collector.addFailure("Unable to connect to Salesforce Instance.",
+ "Ensure properties like Client ID, Client Secret, API Endpoint " +
+ ", Soap Endpoint, Auth Endpoint are correct.")
+ .withConfigProperty(MarketingCloudConstants.PROPERTY_CLIENT_ID)
+ .withConfigProperty(MarketingCloudConstants.PROPERTY_CLIENT_SECRET)
+ .withConfigProperty(MarketingCloudConstants.PROPERTY_API_ENDPOINT)
+ .withConfigProperty(MarketingCloudConstants.PROPERTY_AUTH_API_ENDPOINT)
+ .withConfigProperty(MarketingCloudConstants.PROPERTY_SOAP_API_ENDPOINT)
+ .withStacktrace(e.getStackTrace());
+ }
+ }
+
+ private void validateQueryMode(FailureCollector collector) {
+ //according to query mode check if either object name / object list exists or not
+ if (containsMacro(MarketingCloudConstants.PROPERTY_QUERY_MODE)) {
+ return;
+ }
+
+ SourceQueryMode mode = getQueryMode(collector);
+
+ if (mode == SourceQueryMode.MULTI_OBJECT) {
+ validateMultiObjectQueryMode(collector);
+ } else {
+ validateSingleObjectQueryMode(collector);
+ }
+ }
+
+ private void validateMultiObjectQueryMode(FailureCollector collector) {
+ if (containsMacro(MarketingCloudConstants.PROPERTY_OBJECT_LIST)
+ || containsMacro(MarketingCloudConstants.PROPERTY_DATA_EXTENSION_KEY_LIST)
+ || containsMacro(MarketingCloudConstants.PROPERTY_TABLE_NAME_FIELD)) {
+ return;
+ }
+
+ List objects = getObjectList(collector);
+ collector.getOrThrowException();
+
+ if (objects.isEmpty()) {
+ collector.addFailure("At least 1 Object must be specified.", null)
+ .withConfigProperty(MarketingCloudConstants.PROPERTY_OBJECT_LIST);
+ }
+
+ if (objects.contains(SourceObject.DATA_EXTENSION)) {
+ List dataExtensionKeyList = Util.splitToList(getDataExtensionKeys(), ',');
+ if (dataExtensionKeyList.isEmpty()) {
+ collector.addFailure("At least 1 Data Extension Key must be specified.", null)
+ .withConfigProperty(MarketingCloudConstants.PROPERTY_DATA_EXTENSION_KEY_LIST);
+ }
+ }
+
+ if (Util.isNullOrEmpty(tableNameField)) {
+ collector.addFailure("Table name field must be specified.", null)
+ .withConfigProperty(MarketingCloudConstants.PROPERTY_TABLE_NAME_FIELD);
+ }
+ }
+
+ private void validateSingleObjectQueryMode(FailureCollector collector) {
+ if (containsMacro(MarketingCloudConstants.PROPERTY_OBJECT_NAME)
+ || containsMacro(MarketingCloudConstants.PROPERTY_DATA_EXTENSION_KEY)) {
+ return;
+ }
+
+ SourceObject object = getObject(collector);
+
+ if (object == SourceObject.DATA_EXTENSION && Util.isNullOrEmpty(dataExtensionKey)) {
+ collector.addFailure("Data Extension Key must be specified.", null)
+ .withConfigProperty(MarketingCloudConstants.PROPERTY_DATA_EXTENSION_KEY);
+ }
+ }
+
+ private void validateFilter(FailureCollector collector) {
+ if (containsMacro(MarketingCloudConstants.PROPERTY_FILTER) || Util.isNullOrEmpty(filter)) {
+ return;
+ }
+ try {
+ MarketingCloudClient.validateFilter(filter);
+ } catch (ETSdkException e) {
+ collector.addFailure("Filter string is not valid.",
+ "Check syntax to confirm.")
+ .withConfigProperty(MarketingCloudConstants.PROPERTY_FILTER)
+ .withStacktrace(e.getStackTrace());
+ }
+ }
+
+ /**
+ * Returns true if Salesforce can be connected to.
+ */
+ public boolean shouldConnect() {
+ return !containsMacro(MarketingCloudConstants.PROPERTY_CLIENT_ID) &&
+ !containsMacro(MarketingCloudConstants.PROPERTY_CLIENT_SECRET) &&
+ !containsMacro(MarketingCloudConstants.PROPERTY_API_ENDPOINT) &&
+ !containsMacro(MarketingCloudConstants.PROPERTY_AUTH_API_ENDPOINT) &&
+ !containsMacro(MarketingCloudConstants.PROPERTY_SOAP_API_ENDPOINT);
+ }
+}
diff --git a/src/main/java/io/cdap/plugin/sfmc/source/util/MarketingCloudColumn.java b/src/main/java/io/cdap/plugin/sfmc/source/util/MarketingCloudColumn.java
new file mode 100644
index 0000000..627c29b
--- /dev/null
+++ b/src/main/java/io/cdap/plugin/sfmc/source/util/MarketingCloudColumn.java
@@ -0,0 +1,63 @@
+/*
+ * Copyright © 2020 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.sfmc.source.util;
+
+import com.google.common.base.Strings;
+
+/**
+ * Information about a MarketingCloud table column.
+ */
+public class MarketingCloudColumn {
+ private String fieldName;
+ private String typeName;
+ private String mandatory;
+
+ public MarketingCloudColumn() {
+ }
+
+ /**
+ * Constructor for MarketingCloudColumn.
+ *
+ * @param fieldName The column name
+ * @param typeName The data type name
+ */
+ public MarketingCloudColumn(String fieldName, String typeName) {
+ this.fieldName = fieldName;
+ this.typeName = typeName;
+ this.mandatory = "false";
+ }
+
+ public String getFieldName() {
+ return fieldName;
+ }
+
+ /**
+ * Replaces all space characters in column name with double underscores
+ * @return field name with all characters replaced.
+ */
+ public String getFormattedFieldName() {
+ return fieldName.replaceAll(" ", "__");
+ }
+
+ public String getTypeName() {
+ return typeName;
+ }
+
+ public boolean isMandatory() {
+ return !Strings.isNullOrEmpty(mandatory) && "true".equalsIgnoreCase(mandatory);
+ }
+}
diff --git a/src/main/java/io/cdap/plugin/sfmc/source/util/MarketingCloudConstants.java b/src/main/java/io/cdap/plugin/sfmc/source/util/MarketingCloudConstants.java
new file mode 100644
index 0000000..3b5f953
--- /dev/null
+++ b/src/main/java/io/cdap/plugin/sfmc/source/util/MarketingCloudConstants.java
@@ -0,0 +1,104 @@
+/*
+ * Copyright © 2020 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.sfmc.source.util;
+
+/**
+ * Salesforce constants.
+ */
+public interface MarketingCloudConstants {
+
+ /**
+ * Salesforce plugin name.
+ */
+ String PLUGIN_NAME = "MarketingCloud";
+
+ /**
+ * Configuration property name used to specify the query mode.
+ */
+ String PROPERTY_QUERY_MODE = "queryMode";
+
+ /**
+ * Configuration property name used to specify the object name.
+ */
+ String PROPERTY_OBJECT_NAME = "objectName";
+
+ /**
+ * Configuration property name used to specify data extension key.
+ */
+ String PROPERTY_DATA_EXTENSION_KEY = "dataExtensionKey";
+
+ /**
+ * Configuration property name used to specify the list of objects.
+ */
+ String PROPERTY_OBJECT_LIST = "objectList";
+
+ /**
+ * Configuration property name used to specify the comma-separated list of data extension keys.
+ */
+ String PROPERTY_DATA_EXTENSION_KEY_LIST = "dataExtensionKeyList";
+
+ /**
+ * Configuration property name used to specify the table name field.
+ */
+ String PROPERTY_TABLE_NAME_FIELD = "tableNameField";
+
+ /**
+ * Configuration property name used to specify client id.
+ */
+ String PROPERTY_CLIENT_ID = "clientId";
+
+ /**
+ * Configuration property name used to specify client secret.
+ */
+ String PROPERTY_CLIENT_SECRET = "clientSecret";
+
+ /**
+ * Configuration property name used to specify REST API endpoint.
+ */
+ String PROPERTY_API_ENDPOINT = "restEndpoint";
+
+ /**
+ * Configuration property name used to specify Auth API endpoint.
+ */
+ String PROPERTY_AUTH_API_ENDPOINT = "authEndpoint";
+
+ /**
+ * Configuration property name used to specify SOAP endpoint.
+ */
+ String PROPERTY_SOAP_API_ENDPOINT = "soapEndpoint";
+
+ /**
+ * Configuration property name used to specify Fail on Error flag.
+ */
+ String PROPERTY_FAIL_ON_ERROR = "failOnError";
+
+ /**
+ * Table prefix to be used in case of Multi-Object support.
+ */
+ String TABLE_PREFIX = "multisink.";
+
+ /**
+ * Table name prefix to be used for data extensions.
+ */
+ String DATA_EXTENSION_PREFIX = "dataextension_";
+
+ /**
+ * Filter to be used when retrieving data from Salesforce Marketing Cloud.
+ */
+ String PROPERTY_FILTER = "filter";
+
+}
diff --git a/src/main/java/io/cdap/plugin/sfmc/source/util/MarketingCloudObjectInfo.java b/src/main/java/io/cdap/plugin/sfmc/source/util/MarketingCloudObjectInfo.java
new file mode 100644
index 0000000..20d5846
--- /dev/null
+++ b/src/main/java/io/cdap/plugin/sfmc/source/util/MarketingCloudObjectInfo.java
@@ -0,0 +1,102 @@
+/*
+ * Copyright © 2020 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.sfmc.source.util;
+
+import io.cdap.cdap.api.data.schema.Schema;
+
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * Information about a Salesforce table.
+ */
+public class MarketingCloudObjectInfo {
+ private final SourceObject object;
+ private final String dataExtensionKey;
+ private final Schema schema;
+ private final Map mapSchemaFieldToSFMCField;
+
+
+ /**
+ * Constructor for MarketingCloudObjectInfo for non-dataextension object.
+ * @param object The Salesforce Marketing Cloud object
+ * @param columns The list of columns
+ */
+ public MarketingCloudObjectInfo(SourceObject object, List columns) {
+ this(object, null, columns);
+ }
+
+ /**
+ * Constructor for MarketingCloudObjectInfo for dataextension object.
+ *
+ * @param object The Salesforce Marketing Cloud object as DataExtension
+ * @param dataExtensionKey The data extension key
+ * @param columns The list of columns
+ */
+ public MarketingCloudObjectInfo(SourceObject object, String dataExtensionKey, List columns
+ ) {
+ this.object = object;
+ this.dataExtensionKey = dataExtensionKey;
+ SchemaBuilder schemaBuilder = new SchemaBuilder();
+ this.schema = schemaBuilder.constructSchema(getFormattedTableName(), columns);
+ this.mapSchemaFieldToSFMCField = columns.stream()
+ .collect(Collectors.toMap(MarketingCloudColumn::getFormattedFieldName, MarketingCloudColumn::getFieldName));
+
+ }
+
+ public SourceObject getObject() {
+ return object;
+ }
+
+ /**
+ * Returns the table name for the object.
+ *
+ * @return In case of Data Extension, it returns name in `dataextension-[data extension key]` format
+ * Otherwise, it returns object name
+ */
+ public String getTableName() {
+ if (getObject() == SourceObject.DATA_EXTENSION) {
+ return String.format("%s%s", MarketingCloudConstants.DATA_EXTENSION_PREFIX, dataExtensionKey);
+ } else {
+ return getObject().getTableName();
+ }
+ }
+
+ /**
+ * Replaces all hyphen (-) characters in column name with single underscore
+ * @return table name with all characters replaced.
+ */
+ public String getFormattedTableName() {
+ return getTableName().replaceAll("-", "_");
+ }
+
+ public Schema getSchema() {
+ return schema;
+ }
+
+
+
+ public String lookupFieldsMap(String schemaFieldName) {
+ if (this.mapSchemaFieldToSFMCField == null) {
+ return schemaFieldName;
+ }
+
+ String sfmcFieldName = this.mapSchemaFieldToSFMCField.get(schemaFieldName);
+ return sfmcFieldName == null ? schemaFieldName : sfmcFieldName;
+ }
+}
diff --git a/src/main/java/io/cdap/plugin/sfmc/source/util/SchemaBuilder.java b/src/main/java/io/cdap/plugin/sfmc/source/util/SchemaBuilder.java
new file mode 100644
index 0000000..cd68890
--- /dev/null
+++ b/src/main/java/io/cdap/plugin/sfmc/source/util/SchemaBuilder.java
@@ -0,0 +1,87 @@
+/*
+ * Copyright © 2020 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.sfmc.source.util;
+
+import com.google.common.base.Strings;
+import io.cdap.cdap.api.data.schema.Schema;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+/**
+ * Utility class to build schema.
+ */
+public class SchemaBuilder {
+ /**
+ * Constructs Schema object using input parameters.
+ *
+ * @param tableName The table name to be used in Schema object
+ * @param columns The list of SalesforceColumn objects that will be added as Schema.Field
+ * @return The instance of Schema object
+ */
+ public Schema constructSchema(String tableName, List columns) {
+ SchemaBuilder schemaBuilder = new SchemaBuilder();
+ List fields = schemaBuilder.constructSchemaFields(columns);
+
+ return Schema.recordOf(tableName, fields);
+ }
+
+ private List constructSchemaFields(List columns) {
+ return columns.stream()
+ .map(o -> transformToField(o))
+ .filter(Objects::nonNull)
+ .collect(Collectors.toList());
+ }
+
+ private Schema.Field transformToField(MarketingCloudColumn column) {
+ String name = column.getFormattedFieldName();
+ if (Strings.isNullOrEmpty(name)) {
+ return null;
+ }
+
+ Schema schema = createSchema(column);
+ if (schema == null) {
+ return null;
+ }
+
+ return Schema.Type.NULL == schema.getType()
+ ? Schema.Field.of(name, schema)
+ : Schema.Field.of(name, Schema.nullableOf(schema));
+ }
+
+ private Schema createSchema(MarketingCloudColumn column) {
+ switch (column.getTypeName().toLowerCase()) {
+ case "decimal":
+ case "double":
+ return Schema.of(Schema.Type.DOUBLE);
+ case "number":
+ case "integer":
+ return Schema.of(Schema.Type.INT);
+ case "boolean":
+ return Schema.of(Schema.Type.BOOLEAN);
+ case "date":
+ return Schema.of(Schema.LogicalType.TIMESTAMP_MICROS);
+ case "phone":
+ case "email_address":
+ case "locale":
+ case "string":
+ default:
+ return Schema.of(Schema.Type.STRING);
+ }
+ }
+}
diff --git a/src/main/java/io/cdap/plugin/sfmc/source/util/SourceObject.java b/src/main/java/io/cdap/plugin/sfmc/source/util/SourceObject.java
new file mode 100644
index 0000000..cf07501
--- /dev/null
+++ b/src/main/java/io/cdap/plugin/sfmc/source/util/SourceObject.java
@@ -0,0 +1,138 @@
+/*
+ * Copyright © 2020 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.sfmc.source.util;
+
+import com.custom.fuelsdk.ETNotSentEvent;
+import com.exacttarget.fuelsdk.ETBounceEvent;
+import com.exacttarget.fuelsdk.ETDataExtensionRow;
+import com.exacttarget.fuelsdk.ETEmail;
+import com.exacttarget.fuelsdk.ETList;
+import com.exacttarget.fuelsdk.ETOpenEvent;
+import com.exacttarget.fuelsdk.ETSentEvent;
+import com.exacttarget.fuelsdk.ETSoapObject;
+import com.exacttarget.fuelsdk.ETUnsubEvent;
+
+import java.util.Arrays;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Indicates the object for which data to be fetched from MarketingCloud.
+ */
+public enum SourceObject {
+
+ /**
+ * This indicates data to be fetched from Data Extension.
+ */
+ DATA_EXTENSION("Data Extension", "dataextension", ETDataExtensionRow.class),
+
+ /**
+ * This indicates data to be fetched from Campaign.
+ */
+// CAMPAIGN("Campaign", "campaign", ETCampaign.class),
+
+ /**
+ * This indicates data to be fetched from Email.
+ */
+ EMAIL("Email", "email", ETEmail.class),
+
+ /**
+ * This indicates data to be fetched from Mailing List.
+ */
+ MAILING_LIST("Mailing List", "mailinglist", ETList.class),
+
+ /**
+ * This indicates data to be fetched from Tracking Bounce Events.
+ */
+ TRACKING_BOUNCE_EVENT("Bounce Event", "bounce", ETBounceEvent.class),
+
+ /**
+ * This indicates data to be fetched from Tracking Open Events.
+ */
+ TRACKING_OPEN_EVENT("Open Event", "open", ETOpenEvent.class),
+
+ /**
+ * This indicates data to be fetched from Tracking Click Events.
+ */
+ //TRACKING_CLICK_EVENT("Click Event", "click", ETClickEvent.class),
+
+ /**
+ * This indicates data to be fetched from Tracking UnSub Events.
+ */
+ TRACKING_UNSUB_EVENT("Unsub Event", "unsub", ETUnsubEvent.class),
+
+ /**
+ * This indicates data to be fetched from Tracking Sent Events.
+ */
+ TRACKING_SENT_EVENT("Sent Event", "sent", ETSentEvent.class),
+
+ /**
+ * This indicates data to be fetched from Tracking Notsent Events.
+ */
+ TRACKING_NOTSENT_EVENT("Notsent Event", "notsent", ETNotSentEvent.class);
+
+ private final String value;
+ private final String tableName;
+ private final Class classRef;
+ private String filter = "";
+
+
+ SourceObject(String value, String tableName, Class classRef) {
+ this.value = value;
+ this.tableName = tableName;
+ this.classRef = classRef;
+ }
+
+ /**
+ * Converts object string value into {@link SourceObject} enum.
+ *
+ * @param stringValue object string value
+ * @return source object in optional container
+ */
+ public static Optional fromValue(String stringValue) {
+ return Stream.of(values())
+ .filter(keyType -> keyType.value.equalsIgnoreCase(stringValue))
+ .findAny();
+ }
+
+ public static String getSupportedObjects() {
+ return Arrays.stream(SourceObject.values()).map(SourceObject::getValue)
+ .collect(Collectors.joining(", "));
+ }
+
+ public String getValue() {
+ return value;
+ }
+
+ public String getTableName() {
+ return tableName;
+ }
+
+ public Class getClassRef() {
+ return classRef;
+ }
+
+ public String getFilter() {
+ return filter;
+ }
+
+ public void setFilter(String filter) {
+ this.filter = filter;
+ }
+
+}
diff --git a/src/main/java/io/cdap/plugin/sfmc/source/util/SourceQueryMode.java b/src/main/java/io/cdap/plugin/sfmc/source/util/SourceQueryMode.java
new file mode 100644
index 0000000..f3f228a
--- /dev/null
+++ b/src/main/java/io/cdap/plugin/sfmc/source/util/SourceQueryMode.java
@@ -0,0 +1,65 @@
+/*
+ * Copyright © 2020 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.sfmc.source.util;
+
+import java.util.Arrays;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Indicates query mode which will be used when fetching MarketingCloud data.
+ */
+public enum SourceQueryMode {
+
+ /**
+ * Mode will be used to fetch data for multiple objects.
+ */
+ MULTI_OBJECT("Multi Object"),
+
+ /**
+ * Mode will be used to fetch data for single object.
+ */
+ SINGLE_OBJECT("Single Object");
+
+ private final String value;
+
+ SourceQueryMode(String value) {
+ this.value = value;
+ }
+
+ /**
+ * Converts mode string value into {@link SourceQueryMode} enum.
+ *
+ * @param stringValue mode string value
+ * @return source query mode in optional container
+ */
+ public static Optional fromValue(String stringValue) {
+ return Stream.of(values())
+ .filter(keyType -> keyType.value.equalsIgnoreCase(stringValue))
+ .findAny();
+ }
+
+ public static String getSupportedModes() {
+ return Arrays.stream(SourceQueryMode.values()).map(SourceQueryMode::getValue)
+ .collect(Collectors.joining(", "));
+ }
+
+ public String getValue() {
+ return value;
+ }
+}
diff --git a/src/main/java/io/cdap/plugin/sfmc/source/util/Util.java b/src/main/java/io/cdap/plugin/sfmc/source/util/Util.java
new file mode 100644
index 0000000..b01af08
--- /dev/null
+++ b/src/main/java/io/cdap/plugin/sfmc/source/util/Util.java
@@ -0,0 +1,53 @@
+/*
+ * Copyright © 2019 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.sfmc.source.util;
+
+import com.google.common.base.Splitter;
+import com.google.common.base.Strings;
+
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Utility class.
+ */
+public class Util {
+ /**
+ * Utility function to check if incoming string is empty or not.
+ *
+ * @param string The value to be checked for emptyness
+ * @return true if string is empty otherwise false
+ */
+ public static boolean isNullOrEmpty(String string) {
+ return Strings.isNullOrEmpty(Strings.nullToEmpty(string).trim());
+ }
+
+ /**
+ * Utility function to split incoming string using a delimiter.
+ *
+ * @param value The value to be split
+ * @param delimiter The delimiter to be used for splitting
+ * @return The list of strings
+ */
+ public static List splitToList(String value, char delimiter) {
+ if (isNullOrEmpty(value)) {
+ return Collections.emptyList();
+ }
+
+ return Splitter.on(delimiter).trimResults().splitToList(value);
+ }
+}
diff --git a/src/test/java/io/cdap/plugin/sfmc/source/SalesforceSourceConfigHelper.java b/src/test/java/io/cdap/plugin/sfmc/source/SalesforceSourceConfigHelper.java
new file mode 100644
index 0000000..101ac51
--- /dev/null
+++ b/src/test/java/io/cdap/plugin/sfmc/source/SalesforceSourceConfigHelper.java
@@ -0,0 +1,124 @@
+/*
+ * Copyright © 2019 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.sfmc.source;
+
+/**
+ * Utility class that provides handy methods to construct Salesforce Source Config for testing
+ */
+public class SalesforceSourceConfigHelper {
+
+ public static final String TEST_REF_NAME = "TestRefName";
+ public static final String TEST_CLIENT_ID = "test-client-id";
+ public static final String TEST_CLIENT_SECRET = "test-client-secret";
+ public static final String TEST_REST_ENDPOINT = "TestRestEndpoint";
+ public static final String TEST_AUTH_ENDPOINT = "TestAuthEndpoint";
+ public static final String TEST_SOAP_ENDPOINT = "TestSoapEndpoint";
+ public static final String TEST_OBJECT_NAME = "Data Extension";
+ public static final String TEST_OBJECT_LIST = "Data Extension";
+ private static final String TEST_FILTER = "";
+
+ public static ConfigBuilder newConfigBuilder() {
+ return new ConfigBuilder();
+ }
+
+ public static class ConfigBuilder {
+ private String referenceName = TEST_REF_NAME;
+ private String clientId = TEST_CLIENT_ID;
+ private String clientSecret = TEST_CLIENT_SECRET;
+ private String restEndpoint = TEST_REST_ENDPOINT;
+ private String authEndpoint = TEST_AUTH_ENDPOINT;
+ private String soapEndpoint = TEST_SOAP_ENDPOINT;
+ private String queryMode;
+ private String objectName = TEST_OBJECT_NAME;
+ private String filter = TEST_FILTER;
+ private String dataExtensionKey;
+ private String objectList = TEST_OBJECT_LIST;
+ private String dataExtensionKeys;
+ private String tableNameField = "tablename";
+
+ public ConfigBuilder setReferenceName(String referenceName) {
+ this.referenceName = referenceName;
+ return this;
+ }
+
+ public ConfigBuilder setQueryMode(String queryMode) {
+ this.queryMode = queryMode;
+ return this;
+ }
+
+ public ConfigBuilder setObjectName(String objectName) {
+ this.objectName = objectName;
+ return this;
+ }
+
+ public ConfigBuilder setFilter(String filter) {
+ this.filter = filter;
+ return this;
+ }
+
+ public ConfigBuilder setDataExtensionKey(String dataExtensionKey) {
+ this.dataExtensionKey = dataExtensionKey;
+ return this;
+ }
+
+ public ConfigBuilder setObjectList(String objectList) {
+ this.objectList = objectList;
+ return this;
+ }
+
+ public ConfigBuilder setDataExtensionKeys(String dataExtensionKeys) {
+ this.dataExtensionKeys = dataExtensionKeys;
+ return this;
+ }
+
+ public ConfigBuilder setTableNameField(String tableNameField) {
+ this.tableNameField = tableNameField;
+ return this;
+ }
+
+ public ConfigBuilder setClientId(String clientId) {
+ this.clientId = clientId;
+ return this;
+ }
+
+ public ConfigBuilder setClientSecret(String clientSecret) {
+ this.clientSecret = clientSecret;
+ return this;
+ }
+
+ public ConfigBuilder setRestEndpoint(String restEndpoint) {
+ this.restEndpoint = restEndpoint;
+ return this;
+ }
+
+ public ConfigBuilder setAuthEndpoint(String authEndpoint) {
+ this.authEndpoint = authEndpoint;
+ return this;
+ }
+
+ public ConfigBuilder setSoapEndpoint(String soapEndpoint) {
+ this.soapEndpoint = soapEndpoint;
+ return this;
+ }
+
+ public MarketingCloudSourceConfig build() {
+ return new MarketingCloudSourceConfig(referenceName, queryMode, objectName, dataExtensionKey, objectList,
+ dataExtensionKeys, tableNameField, filter, clientId, clientSecret,
+ restEndpoint, authEndpoint, soapEndpoint, false);
+ }
+ }
+}
diff --git a/src/test/java/io/cdap/plugin/sfmc/source/SalesforceSourceConfigTest.java b/src/test/java/io/cdap/plugin/sfmc/source/SalesforceSourceConfigTest.java
new file mode 100644
index 0000000..f4d732f
--- /dev/null
+++ b/src/test/java/io/cdap/plugin/sfmc/source/SalesforceSourceConfigTest.java
@@ -0,0 +1,274 @@
+/*
+ * Copyright © 2019 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.sfmc.source;
+
+import io.cdap.cdap.etl.api.FailureCollector;
+import io.cdap.cdap.etl.api.validation.CauseAttributes;
+import io.cdap.cdap.etl.api.validation.ValidationException;
+import io.cdap.cdap.etl.mock.validation.MockFailureCollector;
+import io.cdap.plugin.sfmc.source.util.SourceQueryMode;
+import org.junit.Assert;
+import org.junit.Ignore;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.mockito.Mockito;
+
+import static io.cdap.plugin.sfmc.source.SalesforceSourceConfigHelper.TEST_AUTH_ENDPOINT;
+import static io.cdap.plugin.sfmc.source.SalesforceSourceConfigHelper.TEST_CLIENT_ID;
+import static io.cdap.plugin.sfmc.source.SalesforceSourceConfigHelper.TEST_CLIENT_SECRET;
+import static io.cdap.plugin.sfmc.source.SalesforceSourceConfigHelper.TEST_REST_ENDPOINT;
+import static io.cdap.plugin.sfmc.source.SalesforceSourceConfigHelper.TEST_SOAP_ENDPOINT;
+import static io.cdap.plugin.sfmc.source.util.MarketingCloudConstants.PROPERTY_AUTH_API_ENDPOINT;
+import static io.cdap.plugin.sfmc.source.util.MarketingCloudConstants.PROPERTY_CLIENT_ID;
+import static io.cdap.plugin.sfmc.source.util.MarketingCloudConstants.PROPERTY_CLIENT_SECRET;
+import static io.cdap.plugin.sfmc.source.util.MarketingCloudConstants.PROPERTY_DATA_EXTENSION_KEY;
+import static io.cdap.plugin.sfmc.source.util.MarketingCloudConstants.PROPERTY_DATA_EXTENSION_KEY_LIST;
+import static io.cdap.plugin.sfmc.source.util.MarketingCloudConstants.PROPERTY_QUERY_MODE;
+import static io.cdap.plugin.sfmc.source.util.MarketingCloudConstants.PROPERTY_SOAP_API_ENDPOINT;
+import static io.cdap.plugin.sfmc.source.util.MarketingCloudConstants.PROPERTY_TABLE_NAME_FIELD;
+
+/**
+ * Tests for {@link MarketingCloudSourceConfig}.
+ */
+public class SalesforceSourceConfigTest {
+
+ @Rule
+ public ExpectedException thrown = ExpectedException.none();
+
+ @Test
+ public void testQueryModeSingleObject() {
+ SourceQueryMode queryMode = SourceQueryMode.SINGLE_OBJECT;
+ MarketingCloudSourceConfig config = SalesforceSourceConfigHelper.newConfigBuilder()
+ .setQueryMode("Single Object")
+ .build();
+
+ MockFailureCollector collector = new MockFailureCollector();
+ Assert.assertEquals(queryMode, config.getQueryMode(collector));
+ Assert.assertEquals(0, collector.getValidationFailures().size());
+ }
+
+ @Test
+ public void testQueryModeMultiObject() {
+ SourceQueryMode queryMode = SourceQueryMode.MULTI_OBJECT;
+ MarketingCloudSourceConfig config = SalesforceSourceConfigHelper.newConfigBuilder()
+ .setQueryMode("Multi Object")
+ .build();
+
+ MockFailureCollector collector = new MockFailureCollector();
+ Assert.assertEquals(queryMode, config.getQueryMode(collector));
+ Assert.assertEquals(0, collector.getValidationFailures().size());
+ }
+
+ @Test
+ public void testQueryModeInvalid() {
+ MarketingCloudSourceConfig config = SalesforceSourceConfigHelper.newConfigBuilder()
+ .setQueryMode(null)
+ .build();
+
+ try {
+ MockFailureCollector collector = new MockFailureCollector();
+ config.getQueryMode(collector);
+ } catch (ValidationException e) {
+ Assert.assertEquals(1, e.getFailures().size());
+ Assert.assertEquals(PROPERTY_QUERY_MODE, e.getFailures().get(0).getCauses().get(0)
+ .getAttribute(CauseAttributes.STAGE_CONFIG));
+ }
+ }
+
+ @Test
+ public void testValidateClientIdNull() {
+ MockFailureCollector collector = new MockFailureCollector();
+ MarketingCloudSourceConfig config = withSalesforceValidationMock(SalesforceSourceConfigHelper.newConfigBuilder()
+ .setQueryMode("Single Object")
+ .setClientId(null)
+ .setClientSecret(TEST_CLIENT_SECRET)
+ .setRestEndpoint(TEST_REST_ENDPOINT)
+ .setAuthEndpoint(TEST_AUTH_ENDPOINT)
+ .setSoapEndpoint(TEST_SOAP_ENDPOINT)
+ .build(), collector);
+
+ try {
+ config.validate(collector);
+ collector.getOrThrowException();
+ } catch (ValidationException e) {
+ Assert.assertEquals(1, e.getFailures().size());
+ Assert.assertEquals(PROPERTY_CLIENT_ID, e.getFailures().get(0).getCauses().get(0)
+ .getAttribute(CauseAttributes.STAGE_CONFIG));
+ }
+ }
+
+ @Test
+ public void testValidateClientSecretNull() {
+ MockFailureCollector collector = new MockFailureCollector();
+ MarketingCloudSourceConfig config = withSalesforceValidationMock(SalesforceSourceConfigHelper.newConfigBuilder()
+ .setQueryMode("Single Object")
+ .setClientId(TEST_CLIENT_ID)
+ .setClientSecret(null)
+ .setRestEndpoint(TEST_REST_ENDPOINT)
+ .setAuthEndpoint(TEST_AUTH_ENDPOINT)
+ .setSoapEndpoint(TEST_SOAP_ENDPOINT)
+ .build(), collector);
+
+ try {
+ config.validate(collector);
+ collector.getOrThrowException();
+ } catch (ValidationException e) {
+ Assert.assertEquals(1, e.getFailures().size());
+ Assert.assertEquals(PROPERTY_CLIENT_SECRET, e.getFailures().get(0).getCauses().get(0)
+ .getAttribute(CauseAttributes.STAGE_CONFIG));
+ }
+ }
+
+ @Test
+ public void testValidateAuthEndpointNull() {
+ MockFailureCollector collector = new MockFailureCollector();
+ MarketingCloudSourceConfig config = withSalesforceValidationMock(SalesforceSourceConfigHelper.newConfigBuilder()
+ .setQueryMode("Single Object")
+ .setClientId(TEST_CLIENT_ID)
+ .setClientSecret(TEST_CLIENT_SECRET)
+ .setRestEndpoint(TEST_REST_ENDPOINT)
+ .setAuthEndpoint(null)
+ .setSoapEndpoint(TEST_SOAP_ENDPOINT)
+ .build(), collector);
+
+ try {
+ config.validate(collector);
+ collector.getOrThrowException();
+ } catch (ValidationException e) {
+ Assert.assertEquals(1, e.getFailures().size());
+ Assert.assertEquals(PROPERTY_AUTH_API_ENDPOINT, e.getFailures().get(0).getCauses().get(0)
+ .getAttribute(CauseAttributes.STAGE_CONFIG));
+ }
+ }
+
+ @Test
+ public void testValidateSoapEndpointNull() {
+ MockFailureCollector collector = new MockFailureCollector();
+ MarketingCloudSourceConfig config = withSalesforceValidationMock(SalesforceSourceConfigHelper.newConfigBuilder()
+ .setQueryMode("Single Object")
+ .setClientId(TEST_CLIENT_ID)
+ .setClientSecret(TEST_CLIENT_SECRET)
+ .setRestEndpoint(TEST_REST_ENDPOINT)
+ .setAuthEndpoint(TEST_AUTH_ENDPOINT)
+ .setSoapEndpoint(null)
+ .build(), collector);
+
+ try {
+ config.validate(collector);
+ collector.getOrThrowException();
+ } catch (ValidationException e) {
+ Assert.assertEquals(1, e.getFailures().size());
+ Assert.assertEquals(PROPERTY_SOAP_API_ENDPOINT, e.getFailures().get(0).getCauses().get(0)
+ .getAttribute(CauseAttributes.STAGE_CONFIG));
+ }
+ }
+
+ @Test
+ @Ignore
+ public void testValidCredentials() {
+ MockFailureCollector collector = new MockFailureCollector();
+ MarketingCloudSourceConfig config = withSalesforceValidationMock(SalesforceSourceConfigHelper.newConfigBuilder()
+ .setQueryMode("Single Object")
+ .setClientId(TEST_CLIENT_ID)
+ .setClientSecret(TEST_CLIENT_SECRET)
+ .setRestEndpoint(TEST_REST_ENDPOINT)
+ .setAuthEndpoint(TEST_AUTH_ENDPOINT)
+ .setSoapEndpoint(TEST_SOAP_ENDPOINT)
+ .setDataExtensionKey("test-dataextension-key")
+ .build(), collector);
+
+ config.validate(collector);
+
+ Assert.assertEquals(0, collector.getValidationFailures().size());
+ }
+
+ @Test
+ public void testSingleObjectModeMissingDataExtensionKey() {
+ MockFailureCollector collector = new MockFailureCollector();
+ MarketingCloudSourceConfig config = withSalesforceValidationMock(SalesforceSourceConfigHelper.newConfigBuilder()
+ .setQueryMode("Single Object")
+ .setRestEndpoint(TEST_REST_ENDPOINT)
+ .setAuthEndpoint(TEST_AUTH_ENDPOINT)
+ .setSoapEndpoint(TEST_SOAP_ENDPOINT)
+ .setDataExtensionKey(null)
+ .build(), collector);
+
+ try {
+ config.validate(collector);
+ collector.getOrThrowException();
+ } catch (ValidationException e) {
+ Assert.assertEquals(1, e.getFailures().size());
+ Assert.assertEquals(PROPERTY_DATA_EXTENSION_KEY, e.getFailures().get(0).getCauses().get(0)
+ .getAttribute(CauseAttributes.STAGE_CONFIG));
+ }
+ }
+
+ @Test
+ public void testMultiObjectModeMissingDataExtensionKeys() {
+ MockFailureCollector collector = new MockFailureCollector();
+ MarketingCloudSourceConfig config = withSalesforceValidationMock(SalesforceSourceConfigHelper.newConfigBuilder()
+ .setQueryMode("Multi Object")
+ .setClientId(TEST_CLIENT_ID)
+ .setClientSecret(TEST_CLIENT_SECRET)
+ .setRestEndpoint(TEST_REST_ENDPOINT)
+ .setAuthEndpoint(TEST_AUTH_ENDPOINT)
+ .setSoapEndpoint(TEST_SOAP_ENDPOINT)
+ .setDataExtensionKeys(null)
+ .build(), collector);
+
+ try {
+ config.validate(collector);
+ collector.getOrThrowException();
+ } catch (ValidationException e) {
+ Assert.assertEquals(1, e.getFailures().size());
+ Assert.assertEquals(PROPERTY_DATA_EXTENSION_KEY_LIST, e.getFailures().get(0).getCauses().get(0)
+ .getAttribute(CauseAttributes.STAGE_CONFIG));
+ }
+ }
+
+ @Test
+ public void testMultiObjectModeMissingTableNameField() {
+ MockFailureCollector collector = new MockFailureCollector();
+ MarketingCloudSourceConfig config = withSalesforceValidationMock(SalesforceSourceConfigHelper.newConfigBuilder()
+ .setQueryMode("Multi Object")
+ .setClientId(TEST_CLIENT_ID)
+ .setClientSecret(TEST_CLIENT_SECRET)
+ .setRestEndpoint(TEST_REST_ENDPOINT)
+ .setAuthEndpoint(TEST_AUTH_ENDPOINT)
+ .setSoapEndpoint(TEST_SOAP_ENDPOINT)
+ .setDataExtensionKeys("Test-DataExtension-Key1")
+ .setTableNameField(null)
+ .build(), collector);
+
+ try {
+ config.validate(collector);
+ collector.getOrThrowException();
+ } catch (ValidationException e) {
+ Assert.assertEquals(1, e.getFailures().size());
+ Assert.assertEquals(PROPERTY_TABLE_NAME_FIELD, e.getFailures().get(0).getCauses().get(0)
+ .getAttribute(CauseAttributes.STAGE_CONFIG));
+ }
+ }
+
+ private MarketingCloudSourceConfig withSalesforceValidationMock(MarketingCloudSourceConfig config,
+ FailureCollector collector) {
+ MarketingCloudSourceConfig spy = Mockito.spy(config);
+ Mockito.doNothing().when(spy).validateSalesforceConnection(collector);
+ return spy;
+ }
+}
diff --git a/widgets/MarketingCloud-batchsource.json b/widgets/MarketingCloud-batchsource.json
new file mode 100644
index 0000000..358e53f
--- /dev/null
+++ b/widgets/MarketingCloud-batchsource.json
@@ -0,0 +1,284 @@
+{
+ "metadata": {
+ "spec-version": "1.0"
+ },
+ "display-name": "Salesforce Marketing",
+ "configuration-groups": [
+ {
+ "label": "Basic",
+ "properties": [
+ {
+ "widget-type": "textbox",
+ "label": "Reference Name",
+ "name": "referenceName",
+ "widget-attributes": {
+ "placeholder": "Name used to identify this source for lineage"
+ }
+ },
+ {
+ "widget-type": "select",
+ "label": "Data Retrieval Mode",
+ "name": "queryMode",
+ "widget-attributes": {
+ "values": [
+ "Multi Object",
+ "Single Object"
+ ],
+ "default": "Single Object"
+ }
+ }
+ ]
+ },
+ {
+ "label": "Single Object Retrieval",
+ "properties": [
+ {
+ "widget-type": "select",
+ "label": "Object",
+ "name": "objectName",
+ "widget-attributes": {
+ "values": [
+ "Bounce Event",
+ "Data Extension",
+ "Email",
+ "Mailing List",
+ "Notsent Event",
+ "Open Event",
+ "Sent Event",
+ "Unsub Event"
+ ],
+ "default": "Data Extension"
+ },
+ "plugin-function": {
+ "label": "Get Schema",
+ "widget": "outputSchema",
+ "output-property": "schema",
+ "omit-properties": [
+ {
+ "name": "schema"
+ }
+ ]
+ }
+ },
+ {
+ "widget-type": "textbox",
+ "label": "Data Extension External Key",
+ "name": "dataExtensionKey",
+ "widget-attributes": {
+ "placeholder": "External key of the Data Extension from which data to be read."
+ }
+ }
+ ]
+ },
+ {
+ "label": "Multi Object Retrieval",
+ "properties": [
+ {
+ "widget-type": "multi-select",
+ "label": "Object List",
+ "name": "objectList",
+ "widget-attributes": {
+ "options": [
+ {
+ "id": "Bounce Event",
+ "label": "Bounce Event"
+ },
+ {
+ "id": "Data Extension",
+ "label": "Data Extension"
+ },
+ {
+ "id": "Email",
+ "label": "Email"
+ },
+ {
+ "id": "Mailing List",
+ "label": "Mailing List"
+ },
+ {
+ "id": "Notsent Event",
+ "label": "Notsent Event"
+ },
+ {
+ "id": "Open Event",
+ "label": "Open Event"
+ },
+ {
+ "id": "Sent Event",
+ "label": "Sent Event"
+ },
+ {
+ "id": "Unsub Event",
+ "label": "Unsub Event"
+ }
+ ],
+ "delimiter": ",",
+ "default": "Data Extension"
+ }
+ },
+ {
+ "widget-type": "csv",
+ "label": "Data Extension External Keys",
+ "name": "dataExtensionKeyList",
+ "widget-attributes": {
+ "placeholder": "External keys of the Data Extensions from which data to be read."
+ }
+ },
+ {
+ "widget-type": "textbox",
+ "label": "Table Name Field",
+ "name": "tableNameField",
+ "widget-attributes": {
+ "placeholder": "The name of the field that holds the table name.",
+ "default": "tablename"
+ }
+ }
+ ]
+ },
+ {
+ "label": "Filter",
+ "properties": [
+ {
+ "widget-type": "textbox",
+ "label": "filter",
+ "name": "filter",
+ "widget-attributes": {
+ "placeholder": "data filter criteria"
+ }
+ }
+ ]
+ },
+ {
+ "label": "Authentication",
+ "properties": [
+ {
+ "widget-type": "textbox",
+ "label": "Client ID",
+ "name": "clientId",
+ "widget-attributes" : {
+ "placeholder": "Salesforce Marketing Cloud Client ID"
+ }
+ },
+ {
+ "widget-type": "password",
+ "label": "Client Secret",
+ "name": "clientSecret",
+ "widget-attributes" : {
+ "placeholder": "Salesforce Marketing Cloud Client Secret"
+ }
+ },
+ {
+ "widget-type": "textbox",
+ "label": "Authentication Base URI",
+ "name": "authEndpoint",
+ "widget-attributes" : {
+ "placeholder": "Salesforce Auth API Endpoint e.g. https://instance.auth.marketingcloudapis.com"
+ }
+ },
+ {
+ "widget-type": "textbox",
+ "label": "SOAP API Endpoint",
+ "name": "soapEndpoint",
+ "widget-attributes" : {
+ "placeholder": "Salesforce SOAP API Endpoint e.g. https://instance.Salesforce.soap.marketingcloudapis.com/Service.asmx"
+ }
+ },
+ {
+ "widget-type": "textbox",
+ "label": "REST API Base URI",
+ "name": "restEndpoint",
+ "widget-attributes" : {
+ "placeholder": "Salesforce REST API Endpoint e.g. https://instance.rest.marketingcloudapis.com"
+ }
+ }
+ ]
+ },
+ {
+ "label": "Advanced",
+ "properties": [
+ {
+ "widget-type": "radio-group",
+ "name" : "failOnError",
+ "label" : "Error handling",
+ "widget-attributes": {
+ "layout": "inline",
+ "default": "false",
+ "options": [
+ {
+ "id": "true",
+ "label": "Stop on error"
+ },
+ {
+ "id": "false",
+ "label": "Skip on error"
+ }
+ ]
+ }
+ }
+ ]
+ }
+ ],
+ "outputs": [
+ {
+ "name": "schema",
+ "widget-type": "schema"
+ }
+ ],
+ "jump-config": {
+ "datasets": [
+ {
+ "ref-property-name": "referenceName"
+ }
+ ]
+ },
+ "filters": [
+ {
+ "name": "showSingleObject",
+ "condition": {
+ "expression": "queryMode == 'Single Object'"
+ },
+ "show": [
+ {
+ "type": "property",
+ "name": "objectName"
+ },
+ {
+ "type": "property",
+ "name": "dataExtensionKey"
+ }
+ ]
+ },
+ {
+ "name": "showMultiObject",
+ "condition": {
+ "expression": "queryMode == 'Multi Object'"
+ },
+ "show": [
+ {
+ "type": "property",
+ "name": "objectList"
+ },
+ {
+ "type": "property",
+ "name": "dataExtensionKeyList"
+ },
+ {
+ "type": "property",
+ "name": "tableNameField"
+ }
+ ]
+ },
+ {
+ "name": "showSingleObjectDataExtKey",
+ "condition": {
+ "expression": "objectName == 'Data Extension'"
+ },
+ "show": [
+ {
+ "type": "property",
+ "name": "dataExtensionKey"
+ }
+ ]
+ }
+ ]
+}