Fix null pointer exception when importing program members multithread.

.github/workflows/gem-push.yml

@@ -0,0 +1,27 @@
+name: Ruby Gem
+
+on:
+  workflow_dispatch:
+  push:
+    tags:
+      - '*'
+
+jobs:
+  build:
+    name: Build + Publish
+    runs-on: ubuntu-latest
+    permissions:
+      packages: write
+      contents: read
+    steps:
+      - uses: actions/checkout@v2
+      - name: Set up Ruby 2.7
+        uses: ruby/setup-ruby@v1
+        with:
+          ruby-version: 2.7
+      - name: push gem
+        uses: trocco-io/push-gem-to-gpr-action@v1
+        with:
+          language: java
+          gem-path: "./build/gems/*.gem"
+          github-token: "${{ secrets.GITHUB_TOKEN }}"

.gitignore

@@ -11,4 +11,5 @@ build/
 /.metadata/
 .classpath
 .project
-
+bin/
+example.yml

README.md

@@ -14,6 +14,9 @@ embulk-input-marketo is the gem preparing Embulk input plugins for [Marketo](htt
 - Campaign(campaign)
 - Assets Programs (program)
 - Program Members (program_members)
+- List (list)
+- Activity Type (activity_type)
+- Assets Folders (folder)
 
 This plugin uses Marketo REST API.
 
@@ -199,6 +202,50 @@ Get Members by Program Ids or All Program.
 |---------------------|----------|---------------|-----------------------------------------------------------------------------------------------------------------------|
 | **program_ids**     | false    | null          | Import Members by specified Program_ID (comma-separated). If not specified will import all Members by all Program IDs |
 
+### List
+
+List extract all list data from Marketo
+
+`target: list`
+
+Schema type: Static schema
+
+Incremental support: no
+
+Range ingestion: no
+
+### Activity Type
+
+Activity Type extract all activity type data from Marketo
+
+`target: activity_type`
+
+Schema type: Static schema
+
+Incremental support: no
+
+Range ingestion: no
+
+### Assets folders
+
+Get child folders from within a specified root folder or all folders if no root folder is specified.
+
+`target: folder`
+
+Configuration:
+
+| name                | required | default value | description                                                                                                           |
+|---------------------|----------|---------------|-----------------------------------------------------------------------------------------------------------------------|
+| **root_id**         | false    | null          | Parent folder id                                                                                                      |
+| **root_type**       | false    | folder        | Parent folder type, supported values `folder`, `program`                                                              |
+| **max_depth**       | false    | 2             | Maximum folder depth to traverse                                                                                      |
+| **workspace**       | false    | null          | Name of the workspace                                                                                                 |
+
+Schema type: Static schema
+
+Incremental support: no
+
+Range ingestion: no
 
 ## Example
 

build.gradle

@@ -5,6 +5,7 @@ plugins {
     id "jacoco"
     id "signing"
     id "org.embulk.embulk-plugins" version "0.5.5"
+    id "com.palantir.git-version" version "0.12.3"
 }
 
 repositories {
@@ -13,7 +14,20 @@ repositories {
 
 group = "com.treasuredata.embulk.plugins"
 description = "Loads records from Marketo."
-version = "0.6.26"
+version = {
+    def baseVersion = "0.6.26"
+    def troccoVersion = "0.1.1"
+    def tag = "${baseVersion}-trocco-${troccoVersion}"
+    def vd = versionDetails()
+    if (vd.lastTag != "${tag}") {
+        logger.warn "lastTag '${vd.lastTag}' is not '${tag}'"
+    }
+    if (vd.commitDistance == 0 && vd.lastTag ==~ /^[0-9]+\.[0-9]+\.[0-9]+([.-][.a-zA-Z0-9-]+)?/) {
+        vd.lastTag
+    } else {
+        "0.0.0.${vd.gitHash}"
+    }
+}()
 
 sourceCompatibility = 1.8
 targetCompatibility = 1.8
@@ -60,6 +74,11 @@ dependencies {
     implementation "com.sun.xml.bind:jaxb-impl:2.2.11"
     implementation "javax.activation:activation:1.1.1"
 
+    compile 'com.google.guava:guava:18.0'
+    compile  "com.google.code.findbugs:annotations:3.0.1"
+    compile 'org.apache.commons:commons-lang3:3.4'
+    compile 'org.apache.commons:commons-csv:1.8'
+
     testImplementation "junit:junit:4.+"
     testImplementation "org.embulk:embulk-core:$embulkVersion:tests"
     testImplementation "org.embulk:embulk-junit4:$embulkVersion"

gradle/dependency-locks/compileClasspath.lockfile

@@ -17,6 +17,7 @@ javax.xml.bind:jaxb-api:2.2.11
 net.jcip:jcip-annotations:1.0
 org.apache.bval:bval-core:0.5
 org.apache.bval:bval-jsr303:0.5
+org.apache.commons:commons-csv:1.8
 org.apache.commons:commons-lang3:3.12.0
 org.eclipse.jetty:jetty-client:9.4.51.v20230217
 org.eclipse.jetty:jetty-http:9.4.51.v20230217

gradle/dependency-locks/runtimeClasspath.lockfile

@@ -17,6 +17,7 @@ javax.xml.bind:jaxb-api:2.2.11
 net.jcip:jcip-annotations:1.0
 org.apache.bval:bval-core:0.5
 org.apache.bval:bval-jsr303:0.5
+org.apache.commons:commons-csv:1.8
 org.apache.commons:commons-lang3:3.12.0
 org.eclipse.jetty:jetty-client:9.4.51.v20230217
 org.eclipse.jetty:jetty-http:9.4.51.v20230217

src/main/java/org/embulk/input/marketo/CsvTokenizer.java

@@ -3,18 +3,28 @@
 import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonValue;
 import com.google.common.base.Preconditions;
+import org.apache.commons.csv.CSVFormat;
+import org.apache.commons.csv.CSVParser;
 import org.embulk.config.ConfigException;
 import org.embulk.spi.DataException;
 import org.embulk.util.config.Config;
 import org.embulk.util.config.ConfigDefault;
 import org.embulk.util.config.Task;
 import org.embulk.util.text.LineDecoder;
 import org.embulk.util.text.Newline;
+import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.io.Reader;
 import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
 import java.util.ArrayDeque;
 import java.util.ArrayList;
+import java.util.Calendar;
 import java.util.Deque;
 import java.util.List;
 import java.util.Optional;
@@ -24,6 +34,8 @@
  */
 public class CsvTokenizer
 {
+    private static final Logger LOGGER = LoggerFactory.getLogger(CsvTokenizer.class);
+
     enum RecordState
     {
         NOT_END, END,
@@ -130,6 +142,58 @@ public CsvTokenizer(String delimiter, char quote, char escape, String newline, b
         this.nullStringOrNull = nullStringOrNull;
     }
 
+    private Reader inputStream;
+
+    public CsvTokenizer(Reader inputStream)
+    {
+        this.inputStream = inputStream;
+        this.delimiterChar = 0; // Unused
+        this.delimiterFollowingString = null; // Unused
+        this.quote = 0; // Unused
+        this.escape = 0; // Unused
+        this.newline = null; // Unused;
+        this.trimIfNotQuoted = false; // Unused
+        this.maxQuotedSizeLimit = 0; // Unused
+        this.commentLineMarker = null; // Unused
+        this.input = null; // Unused
+        this.nullStringOrNull = null; // Unused
+    }
+
+    public CSVParser csvParse()
+    {
+        try {
+            String path = String.format("tmp_%d.csv", Calendar.getInstance().getTimeInMillis());
+            File file = new File(path);
+            FileWriter filewriter = new FileWriter(file);
+
+            LOGGER.info("create tmp file: " + path);
+
+            BufferedReader b = new BufferedReader(inputStream);
+            String line = b.readLine();
+            int count = 0;
+            while (true) {
+                filewriter.write(line);
+                line = b.readLine();
+                if (line == null) {
+                    break;
+                }
+                filewriter.write("\r\n");
+                count += 1;
+                if (count % 10000 == 0) {
+                    LOGGER.info("import record count: " + count);
+                }
+            }
+            filewriter.close();
+            inputStream.close();
+
+            CSVParser csvParser = CSVParser.parse(file, StandardCharsets.UTF_8, CSVFormat.DEFAULT.withFirstRecordAsHeader());
+            return csvParser;
+        }
+        catch (IOException e) {
+            throw new InvalidValueException(e.getMessage());
+        }
+    }
+
     public long getCurrentLineNumber()
     {
         return lineNumber;
@@ -434,7 +498,7 @@ else if (isDelimiterFollowingFrom(linePos)) {
                     else if (isSpace(c)) {
                         // column has trailing spaces and quoted. TODO should this be rejected?
                     } else {
-                        throw new InvalidValueException(String.format("Unexpected extra character '%c' after a value quoted by '%c'", c, quote));
+                        columnState = ColumnState.QUOTED_VALUE;
                     }
                     break;
 

src/main/java/org/embulk/input/marketo/MarketoInputPluginDelegate.java

@@ -7,11 +7,14 @@
 import org.embulk.base.restclient.RestClientInputPluginDelegate;
 import org.embulk.config.ConfigException;
 import org.embulk.input.marketo.delegate.ActivityBulkExtractInputPlugin;
+import org.embulk.input.marketo.delegate.ActivityTypeInputPlugin;
 import org.embulk.input.marketo.delegate.CampaignInputPlugin;
 import org.embulk.input.marketo.delegate.CustomObjectInputPlugin;
+import org.embulk.input.marketo.delegate.FolderInputPlugin;
 import org.embulk.input.marketo.delegate.LeadBulkExtractInputPlugin;
 import org.embulk.input.marketo.delegate.LeadWithListInputPlugin;
 import org.embulk.input.marketo.delegate.LeadWithProgramInputPlugin;
+import org.embulk.input.marketo.delegate.ListInputPlugin;
 import org.embulk.input.marketo.delegate.ProgramInputPlugin;
 import org.embulk.input.marketo.delegate.ProgramMembersBulkExtractInputPlugin;
 import org.embulk.input.marketo.rest.MarketoRestClient;
@@ -33,7 +36,10 @@ public interface PluginTask
             ProgramInputPlugin.PluginTask,
             MarketoRestClient.PluginTask,
             CustomObjectInputPlugin.PluginTask,
-            ProgramMembersBulkExtractInputPlugin.PluginTask
+            ProgramMembersBulkExtractInputPlugin.PluginTask,
+            ListInputPlugin.PluginTask,
+            ActivityTypeInputPlugin.PluginTask,
+            FolderInputPlugin.PluginTask
     {
         @Config("target")
         Target getTarget();
@@ -79,7 +85,10 @@ public enum Target
         ALL_LEAD_WITH_PROGRAM_ID(new LeadWithProgramInputPlugin()),
         PROGRAM(new ProgramInputPlugin()),
         CUSTOM_OBJECT(new CustomObjectInputPlugin()),
-        PROGRAM_MEMBERS(new ProgramMembersBulkExtractInputPlugin());
+        PROGRAM_MEMBERS(new ProgramMembersBulkExtractInputPlugin()),
+        LIST(new ListInputPlugin()),
+        ACTIVITY_TYPE(new ActivityTypeInputPlugin()),
+        FOLDER(new FolderInputPlugin());
 
         private final RestClientInputPluginDelegate restClientInputPluginDelegate;
 

src/main/java/org/embulk/input/marketo/MarketoService.java

@@ -1,11 +1,13 @@
 package org.embulk.input.marketo;
 
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import org.embulk.input.marketo.delegate.FolderInputPlugin.RootType;
 import org.embulk.input.marketo.model.MarketoField;
 
 import java.io.File;
 import java.util.Date;
 import java.util.List;
+import java.util.Optional;
 import java.util.Set;
 
 /**
@@ -48,4 +50,6 @@ public interface MarketoService
     ObjectNode describeProgramMembers();
 
     File extractProgramMembers(String exportID);
+
+    Iterable<ObjectNode> getFolders(Optional<Long> rootId, RootType rootType, Integer maxDepth, Optional<String> workspace);
 }

src/main/java/org/embulk/input/marketo/MarketoServiceImpl.java

@@ -1,10 +1,12 @@
 package org.embulk.input.marketo;
 
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.base.CaseFormat;
 import com.google.common.base.Function;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Lists;
 import org.apache.commons.lang3.StringUtils;
+import org.embulk.input.marketo.delegate.FolderInputPlugin.RootType;
 import org.embulk.input.marketo.model.BulkExtractRangeHeader;
 import org.embulk.input.marketo.model.MarketoField;
 import org.embulk.input.marketo.rest.MarketoRestClient;
@@ -21,6 +23,7 @@
 import java.io.OutputStream;
 import java.util.Date;
 import java.util.List;
+import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -275,4 +278,12 @@ public InputStream apply(BulkExtractRangeHeader bulkExtractRangeHeader)
             }
         });
     }
+
+    @Override
+    public Iterable<ObjectNode> getFolders(Optional<Long> rootId, RootType rootType, Integer maxDepth, Optional<String> workspace)
+    {
+        String type = CaseFormat.UPPER_UNDERSCORE.to(CaseFormat.UPPER_CAMEL, rootType.name());
+        Optional<String> root = rootId.isPresent() ? Optional.of(String.format("{\"id\": %d, \"type\": \"%s\"}", rootId.get(), type)) : Optional.empty();
+        return marketoRestClient.getFolders(root, maxDepth, workspace);
+    }
 }