Fail the job when validation fails and ms.work.unit.min.units > 0

linkedin · Oct 25, 2021 · f09fc39 · f09fc39
1 parent e99da96
commit f09fc39
Show file tree

Hide file tree

Showing 2 changed files with 30 additions and 1 deletion.
diff --git a/cdi-core/src/main/java/com/linkedin/cdi/configuration/PropertyCollection.java b/cdi-core/src/main/java/com/linkedin/cdi/configuration/PropertyCollection.java
@@ -320,7 +320,33 @@ public Long getMillis(State state) {
     }
   };
 
-  JsonArrayProperties MSTAGE_WATERMARK = new JsonArrayProperties("ms.watermark");
+  JsonArrayProperties MSTAGE_WATERMARK = new JsonArrayProperties("ms.watermark") {
+    @Override
+    public boolean isValid(State state) {
+      final List<String> types = Lists.newArrayList("unit", "datetime");
+      if (super.isValid(state) && !isBlank(state)) {
+        // Derived fields should meet general JsonArray configuration requirements
+        // and contain only JsonObject items that each has a "name" element and a "formula" element
+        JsonArray value = GSON.fromJson(state.getProp(getConfig()), JsonArray.class);
+        if (value.size() == 0) {
+          return false;
+        }
+        if (!value.get(0).isJsonObject()) {
+          return false;
+        }
+
+        if (!value.get(0).getAsJsonObject().has(KEY_WORD_NAME) || !value.get(0).getAsJsonObject().has(KEY_WORD_TYPE)) {
+          return false;
+        }
+
+        String type = value.get(0).getAsJsonObject().get(KEY_WORD_TYPE).getAsString();
+        if (types.stream().noneMatch(t -> t.equalsIgnoreCase(type))) {
+          return false;
+        }
+      }
+      return super.isValid(state);
+    }
+  };
   JsonArrayProperties MSTAGE_WATERMARK_GROUPS = new JsonArrayProperties("ms.watermark.groups");
 
   // default: 0, minimum: 0, maximum: -

diff --git a/cdi-core/src/main/java/com/linkedin/cdi/source/MultistageSource.java b/cdi-core/src/main/java/com/linkedin/cdi/source/MultistageSource.java
@@ -116,6 +116,9 @@ public List<WorkUnit> getWorkunits(SourceState state) {
     jobKeys.logUsage(state);
     if (!jobKeys.validate(state)) {
       LOG.error("Some parameters are invalid, job will do nothing until they are fixed.");
+      if (MSTAGE_WORK_UNIT_MIN_UNITS.get(state) > 0) {
+        throw new RuntimeException(String.format(EXCEPTION_WORK_UNIT_MINIMUM, MSTAGE_WORK_UNIT_MIN_UNITS.get(state), 0));
+      }
       return new ArrayList<>();
     }