Core: Skip testAddManyFilesWithConsistentOrdering if WORKER_THREAD_POOL_SIZE < 3

[ebyhr]

The test fails if WORKER_THREAD_POOL_SIZE < 3.

Expected size: 3 but was: 2 in:
[GenericManifestFile{content=DATA, path=/var/folders/9s/_zwn4r_n2_9bp0krllp1pl3c0000gp/T/junit-2351142152275179508/metadata/4d4b563b-9372-4561-b141-f9434d3e5473-m1.avro, length=1517389, partition_spec_id=0, added_snapshot_id=1, added_data_files_count=15000, added_rows_count=1507481968, existing_data_files_count=0, existing_rows_count=0, deleted_data_files_count=0, deleted_rows_count=0, partitions=[GenericPartitionFieldSummary{contains_null=false, contains_nan=false, lower_bound=[0, 0, 0, 0], upper_bound=[1, 0, 0, 0]}], key_metadata=null, sequence_number=0, min_sequence_number=0, first_row_id=null},
    GenericManifestFile{content=DATA, path=/var/folders/9s/_zwn4r_n2_9bp0krllp1pl3c0000gp/T/junit-2351142152275179508/metadata/4d4b563b-9372-4561-b141-f9434d3e5473-m0.avro, length=1516450, partition_spec_id=0, added_snapshot_id=1, added_data_files_count=15000, added_rows_count=1507482102, existing_data_files_count=0, existing_rows_count=0, deleted_data_files_count=0, deleted_rows_count=0, partitions=[GenericPartitionFieldSummary{contains_null=false, contains_nan=false, lower_bound=[0, 0, 0, 0], upper_bound=[1, 0, 0, 0]}], key_metadata=null, sequence_number=0, min_sequence_number=0, first_row_id=null}]
java.lang.AssertionError: 
Expected size: 3 but was: 2 in:
[GenericManifestFile{content=DATA, path=/var/folders/9s/_zwn4r_n2_9bp0krllp1pl3c0000gp/T/junit-2351142152275179508/metadata/4d4b563b-9372-4561-b141-f9434d3e5473-m1.avro, length=1517389, partition_spec_id=0, added_snapshot_id=1, added_data_files_count=15000, added_rows_count=1507481968, existing_data_files_count=0, existing_rows_count=0, deleted_data_files_count=0, deleted_rows_count=0, partitions=[GenericPartitionFieldSummary{contains_null=false, contains_nan=false, lower_bound=[0, 0, 0, 0], upper_bound=[1, 0, 0, 0]}], key_metadata=null, sequence_number=0, min_sequence_number=0, first_row_id=null},
    GenericManifestFile{content=DATA, path=/var/folders/9s/_zwn4r_n2_9bp0krllp1pl3c0000gp/T/junit-2351142152275179508/metadata/4d4b563b-9372-4561-b141-f9434d3e5473-m0.avro, length=1516450, partition_spec_id=0, added_snapshot_id=1, added_data_files_count=15000, added_rows_count=1507482102, existing_data_files_count=0, existing_rows_count=0, deleted_data_files_count=0, deleted_rows_count=0, partitions=[GenericPartitionFieldSummary{contains_null=false, contains_nan=false, lower_bound=[0, 0, 0, 0], upper_bound=[1, 0, 0, 0]}], key_metadata=null, sequence_number=0, min_sequence_number=0, first_row_id=null}]
	at org.apache.iceberg.TestMergeAppend.testAddManyFilesWithConsistentOrdering(TestMergeAppend.java:110)
	at java.base/java.lang.reflect.Method.invoke(Method.java:580)
	at java.base/java.util.Optional.ifPresent(Optional.java:178)
	at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:184)
	at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197)
	at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:197)
	at java.base/java.util.ArrayList$ArrayListSpliterator.forEachRemaining(ArrayList.java:1708)
	at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:509)
	at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:499)
	at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:151)
	at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:174)
	at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234)
	at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:596)
	at java.base/java.util.ArrayList.forEach(ArrayList.java:1596)
	at java.base/java.util.ArrayList.forEach(ArrayList.java:1596)

[stevenzwu]

I also ran into the same problem.

We should probably fix this problem as part of this PR, which can explicitly control the writer thread pool and parallelism
#16108

cc @dramaticlly

[stevenzwu]

let's just change the assumeThat check to 3 and leave other code unchanged. We can merge this small change quickly to fix the flakiness.

I mentioned the proper long term fix in another comment.

[dramaticlly]

Do we only run into this if the build only have 2 cores available?

If so we can update as Steven suggested as

    int multiplier = 3;
    assumeThat(ThreadPools.WORKER_THREAD_POOL_SIZE).isGreaterThanOrEqualTo(multiplier);

verified with ICEBERG_WORKER_NUM_THREADS=2 ./gradlew :iceberg-core:test --tests "org.apache.iceberg.TestMergeAppend" locally

Also plan to update this test with provided writeManifestWith pool so that we can have deterministic behavior regardless of core available in the build instance

[ebyhr]

Reverted the multiplier change, and just added assumeThat.

[stevenzwu]

merged this. thanks @ebyhr for identifying and fixing the problem, and others for the review.

@dramaticlly we should include the proper fix for this test in the write thread pool PR.