Add stop_on_invalid_record option

README.md

@@ -17,6 +17,7 @@ expand columns having json into multiple columns
   - **name**: name of the column. you can define [JsonPath](http://goessner.net/articles/JsonPath/) style.
   - **type**: type of the column (see below)
   - **format**: format of the timestamp if type is timestamp
+- **stop_on_invalid_record**: Stop bulk load transaction if an invalid record is included (false by default)
 
 ---
 **type of the column**

src/main/java/org/embulk/filter/expand_json/ExpandJsonFilterPlugin.java

@@ -41,6 +41,10 @@ public interface PluginTask
         @Config("time_zone")
         @ConfigDefault("\"UTC\"")
         public String getTimeZone();
+
+        @Config("stop_on_invalid_record")
+        @ConfigDefault("false")
+        boolean getStopOnInvalidRecord();
     }
 
     @Override

src/main/java/org/embulk/filter/expand_json/FilteredPageOutput.java

@@ -13,6 +13,7 @@
 import com.jayway.jsonpath.ReadContext;
 import org.embulk.spi.Column;
 import org.embulk.spi.ColumnConfig;
+import org.embulk.spi.DataException;
 import org.embulk.spi.Exec;
 import org.embulk.spi.Page;
 import org.embulk.spi.PageBuilder;
@@ -106,6 +107,7 @@ public Column getOutputColumn()
 
 
     private final Logger logger = Exec.getLogger(FilteredPageOutput.class);
+    private final boolean stopOnInvalidRecord;
     private final List<UnchangedColumn> unchangedColumns;
     private final List<ExpandedColumn> expandedColumns;
     private final Column jsonColumn;
@@ -188,6 +190,7 @@ private ParseContext initializeParseContext()
 
     FilteredPageOutput(PluginTask task, Schema inputSchema, Schema outputSchema, PageOutput pageOutput)
     {
+        this.stopOnInvalidRecord = task.getStopOnInvalidRecord();
         this.jsonColumn = initializeJsonColumn(task, inputSchema);
         this.unchangedColumns = initializeUnchangedColumns(inputSchema,
                                                            outputSchema,
@@ -203,18 +206,19 @@ private ParseContext initializeParseContext()
     @Override
     public void add(Page page)
     {
-        try {
-            pageReader.setPage(page);
-
-            while (pageReader.nextRecord()) {
+        pageReader.setPage(page);
+        while (pageReader.nextRecord()) {
+            try {
                 setExpandedJsonColumns();
                 setUnchangedColumns();
                 pageBuilder.addRecord();
             }
-        }
-        catch (JsonProcessingException e) {
-            logger.error(e.getMessage());
-            throw Throwables.propagate(e);
+            catch (DataException | JsonProcessingException e) {
+                if (stopOnInvalidRecord) {
+                    throw new DataException(String.format("Found an invalid record"), e);
+                }
+                logger.warn(String.format("Skipped an invalid record (%s)", e.getMessage()));
+            }
         }
     }
 

src/test/java/org/embulk/filter/expand_json/TestExpandJsonFilterPlugin.java

@@ -12,6 +12,7 @@
 import org.embulk.config.ConfigSource;
 import org.embulk.config.TaskSource;
 import org.embulk.spi.Column;
+import org.embulk.spi.DataException;
 import org.embulk.spi.Exec;
 import org.embulk.spi.Page;
 import org.embulk.spi.PageOutput;
@@ -26,15 +27,18 @@
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
+import org.msgpack.value.MapValue;
 import org.msgpack.value.Value;
-import org.msgpack.value.ValueFactory;
+
+import java.util.List;
 
 import static org.embulk.filter.expand_json.ExpandJsonFilterPlugin.Control;
 import static org.embulk.filter.expand_json.ExpandJsonFilterPlugin.PluginTask;
 import static org.embulk.spi.type.Types.*;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
 import static org.msgpack.value.ValueFactory.newArray;
 import static org.msgpack.value.ValueFactory.newBoolean;
 import static org.msgpack.value.ValueFactory.newFloat;
@@ -185,6 +189,73 @@ public void testDefaultValue()
     Expand Test
      */
 
+    @Test
+    public void testStopOnInvalidRecordOption()
+    {
+        String configYaml = "" +
+                "type: expand_json\n" +
+                "json_column_name: _c0\n" +
+                "root: $.\n" +
+                "expanded_columns:\n" +
+                "  - {name: _e0, type: json}\n";
+        final ConfigSource conf = getConfigFromYaml(configYaml);
+        final Schema schema = schema("_c0", STRING);
+
+        { // stop_on_invalid_record: false
+            ConfigSource config = conf.deepCopy();
+
+            expandJsonFilterPlugin.transaction(config, schema, new Control()
+            {
+                @Override
+                public void run(TaskSource taskSource, Schema outputSchema)
+                {
+                    MockPageOutput mockPageOutput = new MockPageOutput();
+
+                    try (PageOutput pageOutput = expandJsonFilterPlugin.open(taskSource, schema, outputSchema, mockPageOutput)) {
+                        for (Page page : PageTestUtils.buildPage(runtime.getBufferAllocator(), schema,
+                                "{\"_e0\":\"\"}", "{\"_e0\":{}}")) {
+                            pageOutput.add(page);
+                        }
+
+                        pageOutput.finish();
+                    }
+
+                    List<Object[]> records = Pages.toObjects(outputSchema, mockPageOutput.pages);
+                    assertEquals(1, records.size());
+                    assertEquals(0, ((MapValue) records.get(0)[0]).size()); // {}
+                }
+            });
+        }
+
+        { // stop_on_invalid_record: true
+            ConfigSource config = conf.deepCopy().set("stop_on_invalid_record", true);
+
+            try {
+                expandJsonFilterPlugin.transaction(config, schema, new Control()
+                {
+                    @Override
+                    public void run(TaskSource taskSource, Schema outputSchema)
+                    {
+                        MockPageOutput mockPageOutput = new MockPageOutput();
+
+                        try (PageOutput pageOutput = expandJsonFilterPlugin.open(taskSource, schema, outputSchema, mockPageOutput)) {
+                            for (Page page : PageTestUtils.buildPage(runtime.getBufferAllocator(), schema,
+                                    "{\"_e0\":\"\"}", "{\"_e0\":{}}")) {
+                                pageOutput.add(page);
+                            }
+
+                            pageOutput.finish();
+                        }
+                    }
+                });
+                fail();
+            }
+            catch (Throwable t) {
+                assertTrue(t instanceof DataException);
+            }
+        }
+    }
+
     @Test
     public void testExpandJsonKeyToSchema()
     {