[inlong] 03/06: [INLONG-8934][Distribution] Add Iceberg connectors into the release bundle (#8935)

[luchunliang]

This is an automated email from the ASF dual-hosted git repository.

luchunliang pushed a commit to branch branch-1.9
in repository https://gitbox.apache.org/repos/asf/inlong.git

commit 97a93f169b41608d96ac37bf539c3427a27f1e7e
Author: vernedeng <verned...@apache.org>
AuthorDate: Tue Sep 19 18:43:46 2023 +0800

    [INLONG-8934][Distribution] Add Iceberg connectors into the release bundle 
(#8935)
---
 inlong-distribution/src/main/assemblies/sort-connectors-v1.15.xml | 8 ++++++++
 .../java/org/apache/inlong/sort/iceberg/source/IcebergSource.java | 4 ----
 .../iceberg/source/reader/InlongIcebergSourceReaderMetrics.java   | 6 ++++--
 3 files changed, 12 insertions(+), 6 deletions(-)

diff --git a/inlong-distribution/src/main/assemblies/sort-connectors-v1.15.xml 
b/inlong-distribution/src/main/assemblies/sort-connectors-v1.15.xml
index db8baf0382..f61fbc0e9e 100644
--- a/inlong-distribution/src/main/assemblies/sort-connectors-v1.15.xml
+++ b/inlong-distribution/src/main/assemblies/sort-connectors-v1.15.xml
@@ -43,5 +43,13 @@
             </includes>
             <fileMode>0644</fileMode>
         </fileSet>
+        <fileSet>
+            
<directory>../inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/target</directory>
+            <outputDirectory>inlong-sort/connectors</outputDirectory>
+            <includes>
+                
<include>sort-connector-iceberg-v1.15-${project.version}.jar</include>
+            </includes>
+            <fileMode>0644</fileMode>
+        </fileSet>
     </fileSets>
 </assembly>
diff --git 
a/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/source/IcebergSource.java
 
b/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/source/IcebergSource.java
index 47f7d04b44..d5b58e050c 100644
--- 
a/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/source/IcebergSource.java
+++ 
b/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/source/IcebergSource.java
@@ -531,10 +531,6 @@ public class IcebergSource<T> implements Source<T, 
IcebergSourceSplit, IcebergEn
             Preconditions.checkNotNull(tableLoader, "tableLoader is 
required.");
             Preconditions.checkNotNull(splitAssignerFactory, "assignerFactory 
is required.");
             Preconditions.checkNotNull(readerFunction, "readerFunction is 
required.");
-            Preconditions.checkNotNull(inlongAuditAddress, "inlongAuditAddress 
is required.");
-            Preconditions.checkNotNull(inlongAuditKeys, "inlongAuditKeys is 
required.");
-            Preconditions.checkNotNull(inlongMetrics, "inlongMetrics is 
required.");
-            Preconditions.checkNotNull(metricOption, "metricOption is 
required.");
         }
 
         private void resolveMetricOption() {
diff --git 
a/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/source/reader/InlongIcebergSourceReaderMetrics.java
 
b/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/source/reader/InlongIcebergSourceReaderMetrics.java
index e7afec2551..60cca86ec3 100644
--- 
a/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/source/reader/InlongIcebergSourceReaderMetrics.java
+++ 
b/inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/source/reader/InlongIcebergSourceReaderMetrics.java
@@ -47,7 +47,9 @@ public class InlongIcebergSourceReaderMetrics extends 
IcebergSourceReaderMetrics
     }
 
     public void outputMetricsWithEstimate(ArrayBatchRecords batchRecord) {
-        sourceMetricData.outputMetricsWithEstimate(batchRecord.records());
-    }
+        if (sourceMetricData != null) {
+            sourceMetricData.outputMetricsWithEstimate(batchRecord.records());
+        }
 
+    }
 }