[Bug] [Iceberg] Iceberg Source use multiple parallelism encountering lost data

[SamealD]

Search before asking

• I had searched in the issues and found no similar issues.

What happened

Iceberg Source use multiple parallelism encountering lost data.
when parallelism=1 ,it won't be lost data. But when parallelism=2 or more ,it will lost data.

SeaTunnel Version

SeaTunnel 2.3.3

SeaTunnel Config

env {
  parallelism = 2
  job.mode = "BATCH"
  checkpoint.interval = 50000
}

source {
  Iceberg {
    catalog_name = "hadoop_prod"
    catalog_type = "hadoop"
    warehouse="hdfs://***:8020/warehouse/hive/test-iceberg"
    namespace = "test01"
    table = "test_table01"
  }
}

sink {
  Console {

 }
}

Running Command

bin/seatunnel.sh --config jobconf/iceberg_to_local.conf

Error Exception

no Error Exception

Zeta or Flink or Spark Version

No response

Java or Scala Version

No response

Screenshots

this is my iceberg table data count:

if I set parallelism = 1, The information I have obtained is as follows:

---

       Job Statistic Information

---

Start Time : 2023-10-19 14:41:10
End Time : 2023-10-19 14:41:16
Total Time(s) : 5
Total Read Count : 2000002
Total Write Count : 2000002
Total Failed Count : 0

---

if I set parallelism = 2, The information I have obtained is as follows:

---

       Job Statistic Information

---

Start Time : 2023-10-19 14:48:58
End Time : 2023-10-19 14:49:01
Total Time(s) : 3
Total Read Count : 1000001
Total Write Count : 1000001
Total Failed Count : 0

---

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[SamealD]

Screenshot failed to upload successfully, It shows original iceberg table data count is 2000002

[kangdw0x80]

It is a bug.
Iceberg Connector assign files to multiple reader with path information (in addPendingSplits function)

• (https://github.com/apache/seatunnel/blob/dev/seatunnel-connectors-v2/connector-iceberg/src/main/java/org/apache/seatunnel/connectors/seatunnel/iceberg/source/enumerator/AbstractSplitEnumerator.java#L110C16-L111C16)

int ownerReader = newSplit.splitId().hashCode() % numReaders;

splitId in Iceberg source use path information.

    public String splitId() {
        return task.file().path().toString();
    }

However, It will get negative value from hashCode function with too long path.

This values is id of reader.
So, the Connector can't assign iceberg file to any reader caused by negative value
Change the code

-   int ownerReader = newSplit.splitId().hashCode() % numReaders;  -> 
+  int ownerReader = ( newSplit.splitId().hashCode() & Integer.MAX_VALUE ) % numReaders;% numReaders;