[Bug] [connector-file] When the data source field is less than the target (Hive) field，it will throw null pointer exception#8150 (#8200)

Author: linjianchang

Diff for: seatunnel-connectors-v2/connector-file/connector-file-base/src/main/java/org/apache/seatunnel/connectors/seatunnel/file/sink/config/FileSinkConfig.java

@@ -187,6 +187,7 @@ public FileSinkConfig(@NonNull Config config, @NonNull SeaTunnelRowType seaTunne
         this.sinkColumnsIndexInRow =
                 this.sinkColumnList.stream()
                         .map(column -> columnsMap.get(column.toLowerCase()))
+                        .filter(e -> e != null)
                         .collect(Collectors.toList());
 
         if (!CollectionUtils.isEmpty(this.partitionFieldList)) {

Diff for: seatunnel-connectors-v2/connector-file/connector-file-base/src/test/java/org/apache/seatunnel/connectors/seatunnel/file/writer/FileSinkConfigTest.java

@@ -31,6 +31,7 @@
 import java.io.File;
 import java.net.URL;
 import java.nio.file.Paths;
+import java.util.List;
 
 public class FileSinkConfigTest {
 
@@ -47,4 +48,23 @@ public void testConfigInit() throws Exception {
                         new SeaTunnelDataType[] {BasicType.STRING_TYPE, BasicType.STRING_TYPE});
         Assertions.assertDoesNotThrow(() -> new FileSinkConfig(config, rowType));
     }
+
+    @Test
+    public void testSinkColumnsGreaterThanSource() throws Exception {
+        URL conf = OrcReadStrategyTest.class.getResource("/test_write_hive.conf");
+        Assertions.assertNotNull(conf);
+        String confPath = Paths.get(conf.toURI()).toString();
+        Config config = ConfigFactory.parseFile(new File(confPath));
+
+        SeaTunnelRowType seaTunnelRowTypeInfo =
+                new SeaTunnelRowType(
+                        new String[] {"name", "age", "address"},
+                        new SeaTunnelDataType[] {
+                            BasicType.STRING_TYPE, BasicType.INT_TYPE, BasicType.STRING_TYPE
+                        });
+        FileSinkConfig fileSinkConfig = new FileSinkConfig(config, seaTunnelRowTypeInfo);
+        List<Integer> sinkColumnsIndexInRow = fileSinkConfig.getSinkColumnsIndexInRow();
+        Assertions.assertEquals(
+                sinkColumnsIndexInRow.size(), seaTunnelRowTypeInfo.getFieldNames().length);
+    }
 }

Diff for: seatunnel-connectors-v2/connector-file/connector-file-base/src/test/resources/test_write_hive.conf

@@ -0,0 +1,25 @@
+#
+# 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.
+#
+
+{
+      fs.defaultFS = "hdfs://hadoop01:9000"
+      path = "/data/test"
+      file_format_type = "json"
+      batch_size=10
+      sink_columns=[name,age,address,weight,height]
+}
+