Fix null pointer exception when importing program members multithread. (

#24)

* handle-null-pointer-exception-when-multithread

* update trocco version

* update comment

build.gradle

@@ -16,7 +16,7 @@ group = "com.treasuredata.embulk.plugins"
 description = "Loads records from Marketo."
 version = {
     def baseVersion = "0.6.26"
-    def troccoVersion = "0.1.0"
+    def troccoVersion = "0.1.1"
     def tag = "${baseVersion}-trocco-${troccoVersion}"
     def vd = versionDetails()
     if (vd.lastTag != "${tag}") {

src/main/java/org/embulk/input/marketo/delegate/ProgramMembersBulkExtractInputPlugin.java

@@ -55,6 +55,7 @@
 public class ProgramMembersBulkExtractInputPlugin extends MarketoBaseInputPluginDelegate<ProgramMembersBulkExtractInputPlugin.PluginTask>
 {
     private final Logger logger = LoggerFactory.getLogger(getClass());
+    private final Object pageBuilderLock = new Object();
 
     public interface PluginTask extends MarketoBaseInputPluginDelegate.PluginTask, CsvTokenizer.PluginTask
     {
@@ -214,7 +215,10 @@ private Future<?> createFutureTask(PluginTask task, RecordImporter recordImporte
                 while (csvRecords.hasNext()) {
                     Map<String, String> csvRecord = csvRecords.next();
                     ObjectNode objectNode = MarketoUtils.OBJECT_MAPPER.valueToTree(csvRecord);
-                    recordImporter.importRecord(new AllStringJacksonServiceRecord(objectNode), pageBuilder);
+                    // MEMO: pageBuilderがスレッドアンセーフなために排他制御を利用する
+                    synchronized (pageBuilderLock) {
+                        recordImporter.importRecord(new AllStringJacksonServiceRecord(objectNode), pageBuilder);
+                    }
                     imported = imported + 1;
                 }