feat(QTDI-1305): improve error in record

component-api/src/main/java/org/talend/sdk/component/api/record/Record.java

@@ -34,6 +34,8 @@
 
 public interface Record {
 
+    String RECORD_ERROR_SUPPORT = "talend.component.record.error.support";
+
     /**
      * @return the schema of this record.
      */
@@ -311,6 +313,11 @@ default Optional<Record> getOptionalRecord(final String name) {
         return ofNullable(get(Record.class, name));
     }
 
+    default boolean isValid() {
+        return !getSchema().getAllEntries()
+                .anyMatch(entry -> !entry.isValid());
+    }
+
     /**
      * Allows to create a record with a fluent API. This is the unique recommended way to create a record.
      */

component-api/src/main/java/org/talend/sdk/component/api/record/Schema.java

@@ -232,6 +232,16 @@ interface Entry {
          */
         boolean isMetadata();
 
+        /**
+         * @return Is this entry can be in error.
+         */
+        boolean isErrorCapable();
+
+        /**
+         * @return true if the value of this entry is valid; false for invalid value.
+         */
+        boolean isValid();
+
         /**
          * @param <T> the default value type.
          *
@@ -296,6 +306,14 @@ default Entry.Builder toBuilder() {
             throw new UnsupportedOperationException("#toBuilder is not implemented");
         }
 
+        default String getErrorMessage() {
+            return getProp(SchemaProperty.ENTRY_ERROR_MESSAGE);
+        }
+
+        default String getErrorFallbackValue() {
+            return getProp(SchemaProperty.ENTRY_ERROR_FALLBACK_VALUE);
+        }
+
         /**
          * Plain builder matching {@link Entry} structure.
          */
@@ -317,6 +335,8 @@ default Builder withLogicalType(String logicalType) {
 
             Builder withNullable(boolean nullable);
 
+            Builder withErrorCapable(boolean errorCapable);
+
             Builder withMetadata(boolean metadata);
 
             <T> Builder withDefaultValue(T value);

component-api/src/main/java/org/talend/sdk/component/api/record/SchemaProperty.java

@@ -37,6 +37,14 @@ public interface SchemaProperty {
 
     String ALLOW_SPECIAL_NAME = "field.special.name";
 
+    String ENTRY_IS_ON_ERROR = "entry.on.error";
+
+    String ENTRY_ERROR_MESSAGE = "entry.error.message";
+
+    String ENTRY_ERROR_FALLBACK_VALUE = "entry.error.fallback.value";
+
+    String ERROR_EXCEPTION = "entry.error.exception";
+
     enum LogicalType {
 
         DATE("date"),

component-api/src/main/java/org/talend/sdk/component/api/service/schema/Schema.java

@@ -129,6 +129,11 @@ public boolean isNullable() {
             return true;
         }
 
+        @Override
+        public boolean isErrorCapable() {
+            return false;
+        }
+
         @Override
         public boolean isMetadata() {
             return false;
@@ -163,6 +168,12 @@ public String getProp(final String property) {
         public Builder toBuilder() {
             throw new UnsupportedOperationException("#toBuilder()");
         }
+
+        @Override
+        public boolean isValid() {
+            return true;
+        }
+
     }
 
 }

component-api/src/test/java/org/talend/sdk/component/api/record/RecordTest.java

@@ -106,6 +106,11 @@ public boolean isNullable() {
                 return false;
             }
 
+            @Override
+            public boolean isErrorCapable() {
+                return false;
+            }
+
             @Override
             public boolean isMetadata() {
                 return false;
@@ -135,6 +140,11 @@ public Map<String, String> getProps() {
             public String getProp(String property) {
                 return null;
             }
+
+            @Override
+            public boolean isValid() {
+                return true;
+            }
         };
         Assertions.assertEquals("value", record.get(String.class, e1));
     }

component-api/src/test/java/org/talend/sdk/component/api/record/SchemaTest.java

@@ -291,6 +291,11 @@ public boolean isNullable() {
             throw new UnsupportedOperationException("#isNullable()");
         }
 
+        @Override
+        public boolean isErrorCapable() {
+            throw new UnsupportedOperationException("#isErrorCapable()");
+        }
+
         @Override
         public boolean isMetadata() {
             throw new UnsupportedOperationException("#isMetadata()");
@@ -326,6 +331,11 @@ public JsonValue getJsonProp(final String name) {
             return Entry.super.getJsonProp(name);
         }
 
+        @Override
+        public boolean isValid() {
+            return true;
+        }
+
     }
 
     class SchemaBuilder implements Schema.Builder {

component-api/src/test/java/org/talend/sdk/component/api/service/record/RecordBuilderFactoryTest.java

@@ -78,6 +78,8 @@ static class MockEntry implements Schema.Entry {
 
         private final boolean nullable;
 
+        private final boolean errorCapable;
+
         private final boolean metadata;
 
         private final Object defaultVal;
@@ -102,6 +104,11 @@ public String getProp(final String property) {
         public Builder toBuilder() {
             throw new UnsupportedOperationException("#toBuilder()");
         }
+
+        @Override
+        public boolean isValid() {
+            return true;
+        }
     }
 
     @RequiredArgsConstructor
@@ -135,6 +142,12 @@ public Entry.Builder withNullable(boolean nullable) {
             return this;
         }
 
+        @Override
+        public Entry.Builder withErrorCapable(boolean errorCapable) {
+            this.builder.withErrorCapable(errorCapable);
+            return this;
+        }
+
         @Override
         public Entry.Builder withMetadata(boolean metadata) {
             this.builder.withMetadata(metadata);

component-runtime-beam/src/test/java/org/talend/sdk/component/runtime/beam/spi/record/AvroRecordBuilderTest.java

@@ -209,6 +209,53 @@ void mixedRecordTest() {
         Assertions.assertNotNull(arrayType);
     }
 
+    @Test
+    void testWithError() {
+        final String val = System.getProperty(Record.RECORD_ERROR_SUPPORT);
+        System.setProperty(Record.RECORD_ERROR_SUPPORT, "true");
+        final String val2 = System.getProperty(Record.RECORD_ERROR_SUPPORT);
+
+        org.talend.sdk.component.api.record.Schema.Builder schemaBuilder = factory.newSchemaBuilder(Schema.Type.RECORD);
+        Schema.Entry nameEntry = factory
+                .newEntryBuilder()
+                .withName("name")
+                .withNullable(false)
+                .withType(Schema.Type.STRING)
+                .build();
+        Schema.Entry nmEntry = factory
+                .newEntryBuilder()
+                .withName("normal")
+                .withNullable(true)
+                .withType(Schema.Type.STRING)
+                .build();
+        Schema.Entry ageEntry = factory
+                .newEntryBuilder()
+                .withName("age")
+                .withNullable(false)
+                .withType(Schema.Type.INT)
+                .build();
+        Schema customerSchema = schemaBuilder.withEntry(nameEntry).withEntry(nmEntry).withEntry(ageEntry).build();
+        // record 1
+        Record.Builder recordBuilder = factory.newRecordBuilder(customerSchema);
+        Record record1 = recordBuilder.with(nameEntry, null)
+                .with(nmEntry, "normal")
+                .with(ageEntry, "is not an int")
+                .build();
+        assertFalse(record1.isValid());
+
+        final Schema.Entry entry =
+                record1.getSchema().getEntries().stream().filter(e -> "name".equals(e.getName())).findAny().get();
+        assertNotNull(entry);
+        Assertions.assertFalse(entry.isValid());
+
+        final Schema.Entry entry2 =
+                record1.getSchema().getEntries().stream().filter(e -> "age".equals(e.getName())).findAny().get();
+        assertNotNull(entry2);
+        Assertions.assertFalse(entry2.isValid());
+
+        System.setProperty(Record.RECORD_ERROR_SUPPORT, "false");
+    }
+
     @Test
     void recordWithNewSchema() {
         final Schema schema0 = new AvroSchemaBuilder()//

component-runtime-beam/src/test/java/org/talend/sdk/component/runtime/beam/spi/record/JsonSchemaSerializationTest.java

@@ -49,7 +49,7 @@ void toJson() throws Exception {
                 .create(new JsonbConfig().withPropertyOrderStrategy(PropertyOrderStrategy.LEXICOGRAPHICAL))) {
             final String json = jsonb.toJson(schema);
             assertEquals(
-                    "{\"entries\":[{\"elementSchema\":{\"entries\":[],\"metadata\":[],\"props\":{},\"type\":\"STRING\"},\"metadata\":false,\"name\":\"array\",\"nullable\":true,\"props\":{\"talend.component.label\":\"array\"},\"rawName\":\"array\",\"type\":\"ARRAY\"}],\"metadata\":[],\"props\":{\"talend.fields.order\":\"array\"},\"type\":\"RECORD\"}",
+                    "{\"entries\":[{\"elementSchema\":{\"entries\":[],\"metadata\":[],\"props\":{},\"type\":\"STRING\"},\"metadata\":false,\"name\":\"array\",\"nullable\":true,\"props\":{\"talend.component.label\":\"array\"},\"rawName\":\"array\",\"type\":\"ARRAY\",\"valid\":true}],\"metadata\":[],\"props\":{\"talend.fields.order\":\"array\"},\"type\":\"RECORD\"}",
                     json);
         }
     }
@@ -71,7 +71,7 @@ void toJsonWithMeta() throws Exception {
         try (final Jsonb jsonb = JsonbBuilder
                 .create(new JsonbConfig().withPropertyOrderStrategy(PropertyOrderStrategy.LEXICOGRAPHICAL))) {
             assertEquals(
-                    "{\"entries\":[],\"metadata\":[{\"elementSchema\":{\"entries\":[],\"metadata\":[],\"props\":{},\"type\":\"STRING\"},\"metadata\":true,\"name\":\"array\",\"nullable\":true,\"props\":{\"talend.component.label\":\"array\"},\"rawName\":\"array\",\"type\":\"ARRAY\"}],\"props\":{\"talend.fields.order\":\"array\"},\"type\":\"RECORD\"}",
+                    "{\"entries\":[],\"metadata\":[{\"elementSchema\":{\"entries\":[],\"metadata\":[],\"props\":{},\"type\":\"STRING\"},\"metadata\":true,\"name\":\"array\",\"nullable\":true,\"props\":{\"talend.component.label\":\"array\"},\"rawName\":\"array\",\"type\":\"ARRAY\",\"valid\":true}],\"props\":{\"talend.fields.order\":\"array\"},\"type\":\"RECORD\"}",
                     jsonb.toJson(schema));
 
         }

component-runtime-impl/src/main/java/org/talend/sdk/component/runtime/record/RecordImpl.java

@@ -61,6 +61,7 @@
 import org.talend.sdk.component.api.record.Schema;
 import org.talend.sdk.component.api.record.Schema.EntriesOrder;
 import org.talend.sdk.component.api.record.Schema.Entry;
+import org.talend.sdk.component.api.record.SchemaProperty;
 
 import lombok.EqualsAndHashCode;
 import lombok.Getter;
@@ -129,6 +130,8 @@ public static class BuilderImpl implements Builder {
 
         private OrderState orderState;
 
+        private Map<String, Entry> entriesInError = new HashMap<>();
+
         public BuilderImpl() {
             this(null);
         }
@@ -150,7 +153,7 @@ private void initOrderState() {
                 } else {
                     final List<Entry> fields = this.providedSchema.naturalOrder()
                             .getFieldsOrder()
-                            .map(this.providedSchema::getEntry)
+                            .map(n -> this.getEntryWithErrorIfAny(this.providedSchema.getEntry(n)))
                             .collect(Collectors.toList());
                     this.orderState = new OrderState(fields);
                 }
@@ -171,11 +174,16 @@ public Object getValue(final String name) {
 
         @Override
         public Builder with(final Entry entry, final Object value) {
-            validateTypeAgainstProvidedSchema(entry.getName(), entry.getType(), value);
+            try {
+                validateTypeAgainstProvidedSchema(entry.getName(), entry.getType(), value);
+            } catch (Exception e) {
+                return withError(entry, value, e.getMessage());
+            }
             if (!entry.getType().isCompatible(value)) {
-                throw new IllegalArgumentException(String
-                        .format("Entry '%s' of type %s is not compatible with value of type '%s'", entry.getName(),
-                                entry.getType(), value.getClass().getName()));
+                return withError(entry, value, String
+                        .format("Entry '%s' of type %s is not compatible with given value of type '%s': '%s'.",
+                                entry.getName(),
+                                entry.getType(), value.getClass().getName(), value));
             }
 
             if (entry.getType() == Schema.Type.DATETIME) {
@@ -323,25 +331,53 @@ private Schema.Entry validateTypeAgainstProvidedSchema(final String name, final
             return entry;
         }
 
+        /**
+         * This method return the updated entry with error information if any.
+         *
+         * @param e The entry to check.
+         * @return The entry updated with error information or the given one.
+         */
+        private Entry getEntryWithErrorIfAny(final Entry e) {
+            if (!e.isErrorCapable()) {
+                // The entry doesn't support error management
+                return e;
+            }
+
+            return entriesInError.getOrDefault(e.getOriginalFieldName(), e);
+        }
+
         public Record build() {
             final Schema currentSchema;
             if (this.providedSchema != null) {
                 final String missing = this.providedSchema
                         .getAllEntries()
+                        .map(this::getEntryWithErrorIfAny)
                         .filter(it -> !it.isNullable() && !values.containsKey(it.getName()))
                         .map(Schema.Entry::getName)
                         .collect(joining(", "));
                 if (!missing.isEmpty()) {
                     throw new IllegalArgumentException("Missing entries: " + missing);
                 }
+
+                Schema schemaWithErrors = this.providedSchema;
+                if (!this.entriesInError.isEmpty()) {
+                    Schema.Builder schemaBuilder = new SchemaImpl.BuilderImpl()
+                            .withType(this.providedSchema.getType());
+                    this.providedSchema.getEntries()
+                            .stream()
+                            .map(this::getEntryWithErrorIfAny)
+                            .forEach(schemaBuilder::withEntry);
+                    schemaWithErrors = schemaBuilder.build();
+                }
+
                 if (orderState != null && orderState.isOverride()) {
-                    currentSchema = this.providedSchema.toBuilder().build(this.orderState.buildComparator());
+                    currentSchema = schemaWithErrors.toBuilder().build(this.orderState.buildComparator());
                 } else {
-                    currentSchema = this.providedSchema;
+                    currentSchema = schemaWithErrors;
                 }
             } else {
                 final Schema.Builder builder = new SchemaImpl.BuilderImpl().withType(RECORD);
-                this.entries.forEachValue(builder::withEntry);
+                this.entries.streams().map(this::getEntryWithErrorIfAny).forEach(builder::withEntry);
                 initOrderState();
                 currentSchema = builder.build(orderState.buildComparator());
             }
@@ -513,6 +549,25 @@ public <T> Builder withArray(final Schema.Entry entry, final Collection<T> value
             return append(entry, values);
         }
 
+        private Builder withError(final Entry entry, final Object value, final String errorMessage) {
+            final boolean supportError = Boolean.parseBoolean(System.getProperty(RECORD_ERROR_SUPPORT, "false"));
+            if (!supportError || !entry.isErrorCapable()) {
+                throw new IllegalArgumentException(errorMessage);
+            } else {
+                // duplicate the schema instance with a modified Entry
+                final Entry updatedEntry = entry.toBuilder()
+                        .withName(entry.getName())
+                        .withNullable(true)
+                        .withType(entry.getType())
+                        .withProp(SchemaProperty.ENTRY_IS_ON_ERROR, "true")
+                        .withProp(SchemaProperty.ENTRY_ERROR_MESSAGE, errorMessage)
+                        .withProp(SchemaProperty.ENTRY_ERROR_FALLBACK_VALUE, String.valueOf(value))
+                        .build();
+                this.entriesInError.put(updatedEntry.getOriginalFieldName(), updatedEntry);
+                return this;
+            }
+        }
+
         private void assertType(final Schema.Type actual, final Schema.Type expected) {
             if (actual != expected) {
                 throw new IllegalArgumentException("Expected entry type: " + expected + ", got: " + actual);