[ADS-3000]: [Java] Support nifi parallelism in NiFi ADB Connector

.github/CODEOWNERS

@@ -0,0 +1,18 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+* @arenadata/cs-java-dev
+/nifi-system-tests/ @arenadata/aqa
+CODEOWNERS @Asmoday @iamlapa @AngryX @Vyacheslav-86

.github/workflows/ci-workflow.yml

@@ -19,7 +19,6 @@ on: [push, pull_request]
 
 env:
   MAVEN_COMMAND: ./mvnw
-  MAVEN_COMMAND_WINDOWS: ./mvnw.cmd
   DEFAULT_MAVEN_OPTS: >-
     -Xmx3g
     -XX:ReservedCodeCacheSize=1g
@@ -237,131 +236,3 @@ jobs:
         if: failure()
       - name: Post Disk Usage
         run: df
-
-  macos-build-jp:
-    timeout-minutes: 150
-    runs-on: macos-latest
-    name: MacOS Zulu JDK 8 JP
-    steps:
-      - name: System Information
-        run: |
-          hostname
-          top -l 1 | grep PhysMem
-          sysctl machdep.cpu
-          df
-      - name: Checkout Code
-        uses: actions/checkout@v3
-      - name: Cache Node Modules
-        uses: actions/cache@v4
-        with:
-          path: |
-            ~/.npm
-            **/node_modules
-          key: ${{ runner.os }}-npm-${{ hashFiles('**/package-lock.json') }}
-      - name: Set up Java 8
-        uses: actions/setup-java@v3
-        with:
-          distribution: 'zulu'
-          java-version: '8'
-          cache: 'maven'
-      - name: Maven Compile
-        env:
-          MAVEN_OPTS: >-
-            ${{ env.COMPILE_MAVEN_OPTS }}
-        run: >
-          ${{ env.MAVEN_COMMAND }}
-          ${{ env.MAVEN_COMPILE_COMMAND }}
-      - name: Maven Verify
-        env:
-          NIFI_CI_LOCALE: >-
-            -Duser.language=ja
-            -Duser.country=JP
-          SUREFIRE_OPTS: >-
-            -Duser.language=ja
-            -Duser.country=JP
-            -Duser.region=JP
-            -Duser.timezone=Asia/Tokyo
-          MAVEN_OPTS: >-
-            ${{ env.DEFAULT_MAVEN_OPTS }}
-            -DargLine=${env.SUREFIRE_OPTS}
-        run: >-
-          ${{ env.MAVEN_COMMAND }}
-          ${{ env.MAVEN_VERIFY_COMMAND }}
-          ${{ env.MAVEN_BUILD_PROFILES }}
-          ${{ env.MAVEN_PROJECTS }}
-      - name: Upload Test Reports
-        uses: actions/upload-artifact@v4
-        with:
-          name: surefire-reports-macos-jp
-          path: |
-            ./**/target/surefire-reports/*.txt
-            ./**/target/surefire-reports/*.xml
-          retention-days: 3
-        if: failure()
-      - name: Post Disk Usage
-        run: df
-
-  windows-build:
-    timeout-minutes: 150
-    runs-on: windows-latest
-    name: Windows Zulu JDK 8 FR
-    steps:
-      - name: System Information
-        run: |
-          systeminfo
-          df
-      - name: Setup Git
-        run:  |
-          git config --global core.autocrlf false
-          git config --global core.longpaths true
-      - name: Checkout Code
-        uses: actions/checkout@v3
-      - name: Cache Node Modules
-        uses: actions/cache@v4
-        with:
-          path: |
-            ~\AppData\npm-cache
-            **\node_modules
-          key: ${{ runner.os }}-npm-${{ hashFiles('**/package-lock.json') }}
-      - name: Set up Java 8
-        uses: actions/setup-java@v3
-        with:
-          distribution: 'zulu'
-          java-version: '8'
-          cache: 'maven'
-      - name: Maven Compile
-        env:
-          MAVEN_OPTS: >-
-            ${{ env.COMPILE_MAVEN_OPTS }}
-        run: >
-          ${{ env.MAVEN_COMMAND_WINDOWS }}
-          ${{ env.MAVEN_COMPILE_COMMAND }}
-      - name: Maven Verify
-        env:
-          NIFI_CI_LOCALE: >-
-            -Duser.language=fr
-            -Duser.country=FR
-          SUREFIRE_OPTS: >-
-            -Duser.language=fr
-            -Duser.country=FR
-            -Duser.region=FR
-            -Duser.timezone=Europe/Paris
-          MAVEN_OPTS: >-
-            ${{ env.DEFAULT_MAVEN_OPTS }}
-            -DargLine=${env.SUREFIRE_OPTS}
-        run: >-
-          ${{ env.MAVEN_COMMAND_WINDOWS }}
-          ${{ env.MAVEN_VERIFY_COMMAND }}
-          ${{ env.MAVEN_BUILD_PROFILES }}
-          ${{ env.MAVEN_PROJECTS }}
-      - name: Upload Test Reports
-        uses: actions/upload-artifact@v4
-        with:
-          name: surefire-reports-windows-fr
-          path: |
-            ./**/target/surefire-reports/*.txt
-            ./**/target/surefire-reports/*.xml
-          retention-days: 3
-        if: failure()
-      - name: Post Disk Usage
-        run: df

nifi-assembly/pom.xml

@@ -1027,7 +1027,7 @@ language governing permissions and limitations under the License. -->
         </dependency>
         <dependency>
             <groupId>org.apache.nifi</groupId>
-            <artifactId>nifi-greenplum-service-nar</artifactId>
+            <artifactId>nifi-greengage-service-nar</artifactId>
             <version>1.28.1</version>
             <type>nar</type>
         </dependency>

nifi-nar-bundles/nifi-standard-services/nifi-gpfdist-service-api/src/main/java/org/apache/nifi/gpfdist/metadata/ColumnDataType.java

@@ -19,5 +19,5 @@
 public interface ColumnDataType {
     String getName();
 
-    GreenplumDataType getType();
+    GreengageDataType getType();
 }

nifi-nar-bundles/nifi-standard-services/nifi-gpfdist-service-api/src/main/java/org/apache/nifi/gpfdist/metadata/ColumnDescription.java

@@ -22,4 +22,6 @@ public interface ColumnDescription {
     ColumnDataType getDataType();
 
     boolean isRequired();
+
+    boolean isNullable();
 }

nifi-nar-bundles/nifi-standard-services/nifi-greenplum-service-bundle/nifi-gpfdist-service/src/main/java/org/apache/nifi/gpfdist/service/context/ContextManager.java → nifi-nar-bundles/nifi-standard-services/nifi-gpfdist-service-api/src/main/java/org/apache/nifi/gpfdist/metadata/ContextManager.java

@@ -11,10 +11,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.apache.nifi.gpfdist.service.context;
-
-import org.apache.nifi.gpfdist.metadata.Context;
-import org.apache.nifi.gpfdist.metadata.ContextId;
+package org.apache.nifi.gpfdist.metadata;
 
 import java.util.Optional;
 

nifi-nar-bundles/nifi-standard-services/nifi-gpfdist-service-api/src/main/java/org/apache/nifi/gpfdist/metadata/GpfidstLoadConfig.java

@@ -0,0 +1,65 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.gpfdist.metadata;
+
+import java.util.Objects;
+
+public class GpfidstLoadConfig {
+    private final long asyncContextTimeoutMs;
+    private final long gpfdistSegmentStreamBufferSize;
+    private final long gpfdistStreamBufferEnqueueTimeoutMs;
+
+    public GpfidstLoadConfig(long asyncContextTimeoutMs,
+                             long gpfdistSegmentStreamBufferSize,
+                             long gpfdistStreamBufferEnqueueTimeoutMs) {
+        this.asyncContextTimeoutMs = asyncContextTimeoutMs;
+        this.gpfdistSegmentStreamBufferSize = gpfdistSegmentStreamBufferSize;
+        this.gpfdistStreamBufferEnqueueTimeoutMs = gpfdistStreamBufferEnqueueTimeoutMs;
+    }
+
+    public long getAsyncContextTimeoutMs() {
+        return asyncContextTimeoutMs;
+    }
+
+    public long getGpfdistSegmentStreamBufferSize() {
+        return gpfdistSegmentStreamBufferSize;
+    }
+
+    public long getGpfdistStreamBufferEnqueueTimeoutMs() {
+        return gpfdistStreamBufferEnqueueTimeoutMs;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) return false;
+        GpfidstLoadConfig that = (GpfidstLoadConfig) o;
+        return asyncContextTimeoutMs == that.asyncContextTimeoutMs
+                && gpfdistSegmentStreamBufferSize == that.gpfdistSegmentStreamBufferSize
+                && gpfdistStreamBufferEnqueueTimeoutMs == that.gpfdistStreamBufferEnqueueTimeoutMs;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(asyncContextTimeoutMs, gpfdistSegmentStreamBufferSize, gpfdistStreamBufferEnqueueTimeoutMs);
+    }
+
+    @Override
+    public String toString() {
+        return "GpfidstLoadConfig{" +
+                "asyncContextTimeoutMs=" + asyncContextTimeoutMs +
+                ", gpfdistSegmentStreamBufferSize=" + gpfdistSegmentStreamBufferSize +
+                ", gpfdistStreamBufferEnqueueTimeoutMs=" + gpfdistStreamBufferEnqueueTimeoutMs +
+                '}';
+    }
+}

nifi-nar-bundles/nifi-standard-services/nifi-gpfdist-service-api/src/main/java/org/apache/nifi/gpfdist/metadata/GreenplumDataType.java → nifi-nar-bundles/nifi-standard-services/nifi-gpfdist-service-api/src/main/java/org/apache/nifi/gpfdist/metadata/GreengageDataType.java

@@ -13,7 +13,7 @@
  */
 package org.apache.nifi.gpfdist.metadata;
 
-public enum GreenplumDataType {
+public enum GreengageDataType {
     BOOLEAN,
     MONEY,
     UUID,
@@ -34,4 +34,5 @@ public enum GreenplumDataType {
     TIMESTAMP_WITHOUT_TIME_ZONE,
     ARRAY,
     MAP,
+    ENUM
 }

nifi-nar-bundles/nifi-standard-services/nifi-gpfdist-service-api/src/main/java/org/apache/nifi/gpfdist/metadata/RecordProcessorId.java

@@ -0,0 +1,61 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.nifi.gpfdist.metadata;
+
+import java.util.Objects;
+
+public class RecordProcessorId {
+    private final String sinkId;
+    private final String txId;
+    private final int segmentId;
+
+    public RecordProcessorId(String sinkId, String txId, int segmentId) {
+        this.sinkId = sinkId;
+        this.txId = txId;
+        this.segmentId = segmentId;
+    }
+
+    public String getSinkId() {
+        return sinkId;
+    }
+
+    public String getTxId() {
+        return txId;
+    }
+
+    public int getSegmentId() {
+        return segmentId;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (o == null || getClass() != o.getClass()) return false;
+        RecordProcessorId that = (RecordProcessorId) o;
+        return segmentId == that.segmentId && Objects.equals(sinkId, that.sinkId) && Objects.equals(txId, that.txId);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(sinkId, txId, segmentId);
+    }
+
+    @Override
+    public String toString() {
+        return "RecordProcessorId{" +
+                "sinkId='" + sinkId + '\'' +
+                ", txId='" + txId + '\'' +
+                ", segmentId=" + segmentId +
+                '}';
+    }
+}