[Fix][Connector-V2] Fix possible data loss in scenarios of request_tablet_size is less than the number of BUCKETS

[xiaochen-zhou]

Purpose of this pull request

When users explicitly set QUERY_TABLET_SIZE instead of using the default value Integer.MAX_VALUE, the returned List partitions contains partitions with the same beAddress.

List<StarRocksSourceSplit> getStarRocksSourceSplit() {
  List<StarRocksSourceSplit> sourceSplits = new ArrayList<>();
  List<QueryPartition> partitions = starRocksQueryPlanReadClient.findPartitions();
  for (int i = 0; i < partitions.size(); i++) {
    sourceSplits.add(
      new StarRocksSourceSplit(
        partitions.get(i), String.valueOf(partitions.get(i).hashCode())));
  }
  return sourceSplits;
}

To avoid establishing a connection to the BE for each split during the read process, the StarRocksBeReadClient is cached based on the beAddress from the split's partition. However, in StarRocksBeReadClient#openScanner(), some variables are not being reset, leading to the potential use of stale values from the cache. For example, if eos (end of stream) is true from a previous partition read, it can cause data loss for new partitions on the same BE.

Does this PR introduce any user-facing change?

no

How was this patch tested?

add new tests

Check list

• If any new Jar binary package adding in your PR, please add License Notice according
  New License Guide
• If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
• If you are contributing the connector code, please check that the following files are updated:
  1. Update plugin-mapping.properties and add new connector information in it
  2. Update the pom file of seatunnel-dist
  3. Add ci label in label-scope-conf
  4. Add e2e testcase in seatunnel-e2e
  5. Update connector plugin_config
• Update the release-note.

[xiaochen-zhou]

Friendly ping, do you have time to take a look @Hisoka-X 🙏 ?

[hailin0]

Please add testcase (reproduce & verify)

[xiaochen-zhou]

Please add testcase (reproduce & verify)

Ok. I'll try to complete it today.

[xiaochen-zhou]

I added a test case, StarRocksIT#testStarRocksReadRowCount(), to verify whether the number of rows written to the sink matches the number of rows read from the source in scenarios where request_tablet_size is less than the number of BUCKETS.

When I set the table's buckets to 3:

DISTRIBUTED BY HASH(`BIGINT_COL`) BUCKETS 3

At the same time, when request_tablet_size is set to a value less than 3:

The StarRocksIT#testStarRocksReadRowCount() test could not pass before the fix:

In this case, the row count is 31, which is less than the expected 100.

After applying the fix, the StarRocksIT#testStarRocksReadRowCount() test now passes successfully:

@hailin0 @Hisoka-X

[Hisoka-X]

Thanks @xiaochen-zhou

[hailin0]

good pr