Merge latest code

Author: Chris Li

README.md

@@ -17,17 +17,32 @@ LinkedIn Data Integration Library (DIL) is a collection of generic data integrat
 - Ingest data from one Rest API and egress to another (Rest API) on cloud
 
 # Requirements
-* Java >= 1.8
+* JDK 1.8
 
 If building the distribution with tests turned on:
 * Maven version 3.5.3 
 
 # Instructions to build the distribution
 1. Extract the archive file to your local directory.
-2. Set JAVA_HOME
+2. Set JAVA_HOME to use JDK 1.8 (JDK 11+ not supported)
 3. Build
-* Skip tests and build the distribution
-> `./gradlew build -x findbugsMain -x test -x rat -x checkstyleMain` 
-
-* Tests and build the distribution (requires Maven): 
 > `./gradlew build` 
+
+# Instructions to contribute 
+To contribute, please use submit Pull Request (PR) for committers to merge. 
+- Create your own fork on GitHub off the main repository
+- Clone your fork to your local computer
+>- `git clone https://github.com/<<your-github-login>>/data-integration-library.git`
+- Add upstream and verify
+>- `git remote add upstream https://github.com/linkedin/data-integration-library.git`
+>- `git remote -v`
+- Change, test, commit, and push to your fork
+>- `git status`
+>- `git add .`
+>- `git commit -m "comments"`
+>- `git push origin master`
+- Create Pull Request on GitHub with the following details
+>- Title 
+>- Detailed description
+>- Document the tests done
+>- Links to the updated documents

buildSrc/src/main/groovy/org/apache/gobblin/gradle/BuildProperties.groovy

@@ -7,11 +7,11 @@
 import com.google.gson.Gson;
 import com.google.gson.JsonArray;
 import com.google.gson.JsonObject;
+import com.linkedin.cdi.factory.DefaultConnectionClientFactory;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang.StringUtils;
 import org.apache.gobblin.configuration.State;
-import com.linkedin.cdi.factory.DefaultS3ClientFactory;
 
 
 /**
@@ -117,7 +117,20 @@ public <T> T getDefaultValue() {
    * this value is in milliseconds.
    */
   MSTAGE_CALL_INTERVAL("ms.call.interval.millis", Long.class),
+  MSTAGE_CONVERTER_CSV_MAX_FAILURES("ms.converter.csv.max.failures", Long.class),
+  MSTAGE_CONVERTER_KEEP_NULL_STRINGS("ms.converter.keep.null.strings", Boolean.class) {
+    @Override
+    public <T> T getDefaultValue() {
+      return (T) Boolean.FALSE;
+    }
+  },
   MSTAGE_CSV_COLUMN_HEADER("ms.csv.column.header", Boolean.class),
+  MSTAGE_CSV_COLUMN_HEADER_INDEX("ms.csv.column.header.index", Integer.class) {
+    @Override
+    public <T> T getDefaultValue() {
+      return (T) Integer.valueOf(0);
+    }
+  },
   /**
    * a comma-separated string, where each value is either an integer or a range
    * representing the index of the field to include
@@ -155,6 +168,18 @@ public <T> T getDefaultValue() {
       return (T) ",";
     }
   },
+  /**
+   * By default, CsvExtractor tries to infer the true type of fields when inferring schema
+   * However, in some cases, the inference is not accurate, and users may prefer to keep all fields as strings.
+   * In this case ms.csv.default.field.type = string
+   * Supported types: string | int | long | double | boolean | float
+   */
+  MSTAGE_CSV_DEFAULT_FIELD_TYPE("ms.csv.default.field.type", String.class)  {
+    @Override
+    public <T> T getDefaultValue() {
+      return (T) StringUtils.EMPTY;
+    }
+  },
   /**
    * if csv.column.header is true, csv.skip.lines will be 1 by default, if more than 1
    * row to be skipped, then set this parameter explicitly.
@@ -332,13 +357,13 @@ public Long getMillis(State state) {
     }
   },
   /**
-   * http.client.factory define an indirect way to specify the type of HttpClient to use.
-   * default = {@link com.linkedin.cdi.factory.ApacheHttpClientFactory}
+   * Define an indirect way to specify the type of connection clients
+   * default = {@link DefaultConnectionClientFactory}
    */
-  MSTAGE_HTTP_CLIENT_FACTORY("ms.http.client.factory", String.class) {
+  MSTAGE_CONNECTION_CLIENT_FACTORY("ms.connection.client.factory", String.class) {
     @Override
     public <T> T getDefaultValue() {
-      return (T) "com.linkedin.cdi.factory.ApacheHttpClientFactory";
+      return (T) "com.linkedin.cdi.factory.DefaultConnectionClientFactory";
     }
   },
   /**
@@ -387,17 +412,6 @@ public <T> T getDefaultValue() {
    * Currently, we don't allow exceptions being made to revert errors by using reason code.
    */
   MSTAGE_HTTP_STATUS_REASONS("ms.http.status.reasons", JsonObject.class),
-  /**
-   * jdbc.client.factory define an indirect way to specify the type of JDBC Client to use.
-   * default = {@link com.linkedin.cdi.factory.DefaultJdbcClientFactory}
-   */
-  MSTAGE_JDBC_CLIENT_FACTORY("ms.jdbc.client.factory", String.class) {
-    @Override
-    public <T> T getDefaultValue() {
-      return (T) "com.linkedin.cdi.factory.DefaultJdbcClientFactory";
-    }
-  },
-
   MSTAGE_JDBC_SCHEMA_REFACTOR("ms.jdbc.schema.refactor", String.class) {
     @Override
     public <T> T getDefaultValue() {
@@ -497,6 +511,24 @@ public boolean validateNonblank(State state) {
    * path and fields, etc.
    */
   MSTAGE_PAYLOAD_PROPERTY("ms.payload.property", JsonArray.class),
+  /**
+   * This property is required for inflowValidation with simple count comparison rule
+   * The rule is accepted as a JsonObject with following Keys
+   * 1. "threshold" - represents the percentage of accepted failure records to mark job as passed
+   * 2. "criteria" - this value can be "fail" or "success" , fail represents that input record only has failed records
+   * 3. "errorColumn" - this value is optional and is required when we require to filter the failure records based on a specific column
+   * if the input record has only success records then set this as "success"
+   * Ex: ms.validation.attributes={"threshold": "10", "criteria" : "fail"}
+   */
+  MSTAGE_VALIDATION_ATTRIBUTES("ms.validation.attributes", JsonObject.class) {
+    @Override
+    public <T> T getDefaultValue() {
+      JsonObject attributesJson = new JsonObject();
+      attributesJson.addProperty(StaticConstants.KEY_WORD_THRESHOLD, 0);
+      attributesJson.addProperty(StaticConstants.KEY_WORD_CRITERIA, StaticConstants.KEY_WORD_FAIL);
+      return (T) attributesJson;
+    }
+  },
   MSTAGE_RETENTION("ms.retention", JsonObject.class) {
     @Override
     public <T> T getDefaultValue() {
@@ -507,16 +539,6 @@ public <T> T getDefaultValue() {
       return (T) retention;
     }
   },
-  /**
-   * s3.client.factory define an indirect way to specify the type of S3 Client to use.
-   * default = {@link DefaultS3ClientFactory}
-   */
-  MSTAGE_S3_CLIENT_FACTORY("ms.s3.client.factory", String.class) {
-    @Override
-    public <T> T getDefaultValue() {
-      return (T) "com.linkedin.cdi.factory.DefaultS3ClientFactory";
-    }
-  },
   /**
    * Schema cleansing will replace special characters in the schema element names based
    * on a pattern. By default it will replace all blank spaces, $, and @ to underscores.
@@ -733,6 +755,27 @@ public <T> T getDefaultValue() {
    */
   MSTAGE_WATERMARK("ms.watermark", JsonArray.class),
   MSTAGE_WATERMARK_GROUPS("ms.watermark.groups", JsonArray.class),
+  /**
+   * Minimum records to be present in order for the work unit to be successful,
+   * below the minimum value, the work unit will be failed.
+   */
+  MSTAGE_WORK_UNIT_MIN_RECORDS("ms.work.unit.min.records", Long.class) {
+    @Override
+    public <T> T getDefaultValue() {
+      return (T) Long.valueOf(0);
+    }
+  },
+  /**
+   * Minimum number of work units to be present in order for the job to proceed,
+   * below the minimum value, the job will be failed. This parameter shold be used
+   * only when there is a unit watermark.
+   */
+  MSTAGE_WORK_UNIT_MIN_UNITS("ms.work.unit.min.units", Long.class) {
+    @Override
+    public <T> T getDefaultValue() {
+      return (T) Long.valueOf(0);
+    }
+  },
   MSTAGE_WORK_UNIT_PARALLELISM_MAX("ms.work.unit.parallelism.max", Integer.class) {
     @Override
     public boolean validateNonblank(State state) {
@@ -799,6 +842,16 @@ public <T> T getDefaultValue() {
       return (T) Long.valueOf(500L);
     }
   },
+  MSTAGE_AUDIT_ENABLED("ms.audit.enabled", Boolean.class) {
+    @Override
+    public <T> T getDefaultValue() {
+      return (T) Boolean.FALSE;
+    }
+  },
+  MSTAGE_KAFKA_BROKERS("ms.kafka.brokers", String.class),
+  MSTAGE_KAFKA_SCHEMA_REGISTRY_URL("ms.kafka.schema.registry.url", String.class),
+  MSTAGE_KAFKA_CLIENT_ID("ms.kafka.clientId", String.class),
+  MSTAGE_KAFKA_TOPIC_NAME("ms.kafka.audit.topic.name", String.class),
   // Properties defined in Gobblin, redefine here to leverage the new features like validation
   CONVERTER_CLASSES("converter.classes", String.class),
   DATASET_URN_KEY("dataset.urn", String.class),

buildSrc/src/main/groovy/org/apache/gobblin/gradle/BuildProperty.groovy

@@ -19,10 +19,12 @@ public interface StaticConstants {
   String KEY_WORD_DATA_TYPE = "dataType";
   String KEY_WORD_DATA_TYPE_TYPE = "dataType.type";
   String KEY_WORD_EOF = "EOF";
+  String KEY_WORD_EPOC = "epoc";
   String KEY_WORD_FIELDS = "fields";
   String KEY_WORD_RANGE_FROM = "from";
   String KEY_WORD_HTTP_OK = "ok";
   String KEY_WORD_INTEGER = "integer";
+  String KEY_WORD_IS_NULLABLE = "isNullable";
   String KEY_WORD_ITEMS = "items";
   String KEY_WORD_MAP = "map";
   String KEY_WORD_NAME = "name";
@@ -37,6 +39,7 @@ public interface StaticConstants {
   String KEY_WORD_PROPERTIES = "properties";
   String KEY_WORD_RANGE = "range";
   String KEY_WORD_RECORD = "record";
+  String KEY_WORD_REGEXP = "regexp";
   String KEY_WORD_RETRY = "retry";
   String KEY_WORD_RETRY_COUNT = "retryCount";
   String KEY_WORD_RETRY_DELAY_IN_SEC = "delayInSec";
@@ -52,7 +55,25 @@ public interface StaticConstants {
   String KEY_WORD_UNITS = "units";
   String KEY_WORD_UNKNOWN = "unknown";
   String KEY_WORD_VALUES = "values";
+  String KEY_WORD_THRESHOLD = "threshold";
+  String KEY_WORD_CRITERIA = "criteria";
+  String KEY_WORD_FAIL = "fail";
+  String KEY_WORD_SUCCESS = "success";
+  String KEY_WORD_ERROR_COLUMN = "errorColumn";
+  String KEY_WORD_INT = "int";
+  String KEY_WORD_LONG = "long";
+  String KEY_WORD_DOUBLE = "double";
+  String KEY_WORD_FLOAT = "float";
+  String KEY_WORD_JSON = "json";
+  String KEY_WORD_CSV = "csv";
+  String KEY_WORD_AVRO = "avro";
 
-  Gson GSON = new Gson();
+  String EXCEPTION_WORK_UNIT_MINIMUM = "Job requires a minimum of %s work unit(s) to proceed because ms.work.unit.min.units = %s.";
+  String EXCEPTION_RECORD_MINIMUM = "Work unit requires a minimum of %s record(s) to succeed because ms.work.unit.min.records = %s.";
 
+  String MSG_ROWS_PROCESSED = "Processed %s records, work unit: %s";
+  String MSG_WORK_UNIT_ALWAYS = "There should be a work unit.";
+  String MSG_LOW_WATER_MARK_ALWAYS = "There should be a low watermark.";
+  String MSG_WORK_UNIT_INFO = "Generating Work Unit: %s, watermark: %s";
+  Gson GSON = new Gson();
 }

cdi-core/src/main/java/com/linkedin/cdi/configuration/MultistageProperties.java

@@ -5,6 +5,7 @@
 package com.linkedin.cdi.connection;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
 import java.io.InputStream;
 import java.net.URI;
@@ -66,7 +67,7 @@ public HdfsConnection(State state, JobKeys jobKeys, ExtractorKeys extractorKeys)
    */
   @Override
   public WorkUnitStatus execute(final WorkUnitStatus status) {
-    assert hdfsKeys.getSourceUri() != null;
+    Preconditions.checkNotNull(hdfsKeys.getSourceUri(), "ms.source.uri is missing or of wrong format");
     URI uri = URI.create(getWorkUnitSpecificString(hdfsKeys.getSourceUri(),
         getExtractorKeys().getDynamicParameters()));
 
@@ -142,6 +143,7 @@ private List<String> readFileList(final String path, final String pattern) {
    * @return the file content in an InputStream
    */
   private InputStream readSingleFile(final String path) {
+    log.info("Processing file: {}", path);
     try {
       return fsHelper.getFileStream(path);
     } catch (FileBasedHelperException e) {