[jira] [Created] (FLINK-40257) MongoDB CDC SampleBucketSplitStrategy calculates negative $sample size when collection count exceeds Integer.MAX_VALUE

["soyeong choe (Jira)" <ji...@apache.org>]

soyeong choe created FLINK-40257:
------------------------------------

             Summary: MongoDB CDC SampleBucketSplitStrategy calculates negative $sample size when collection count exceeds Integer.MAX_VALUE
                 Key: FLINK-40257
                 URL: https://issues.apache.org/jira/browse/FLINK-40257
             Project: Flink
          Issue Type: Bug
          Components: Flink CDC
    Affects Versions: cdc-3.6.0
         Environment: * Flink CDC version: 3.6.0
* Connector: MongoDB CDC
* scan.incremental.snapshot.enabled: true
* scan.incremental.snapshot.chunk.samples: 20
* splitVector privilege: not granted
* MongoDB collection document count: > Integer.MAX_VALUE
            Reporter: soyeong choe


h3. Problem

In the MongoDB CDC connector, when the splitVector command is unauthorized or fails, the connector falls back to SampleBucketSplitStrategy.

For very large MongoDB collections whose document count exceeds Integer.MAX_VALUE, SampleBucketSplitStrategy calculates a negative sample size due to integer overflow. This negative value is then passed to MongoDB $sample, causing the snapshot split generation to fail.
h3. Affected code

Current master:

[https://github.com/apache/flink-cdc/blob/master/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/assigners/splitters/SampleBucketSplitStrategy.java#L90-L97]

Release 3.6.0:

[https://github.com/apache/flink-cdc/blob/release-3.6.0/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/main/java/org/apache/flink/cdc/connectors/mongodb/source/assigners/splitters/SampleBucketSplitStrategy.java#L90-L97]

Current code:
{code:java}
long count = splitContext.getDocumentCount();

int numberOfSamples;
if (count < DEFAULT_SAMPLING_THRESHOLD) {
    numberOfSamples = (int) count;
} else {
    numberOfSamples = Math.min(numChunks * splitContext.getSamplesPerChunk(), (int) count);
}
{code}
The value count is a long, but it is cast to int before calculating numberOfSamples. When count is greater than Integer.MAX_VALUE, this cast can overflow to a negative value.
h3. Example

For example, a collection count around 3,727,914,087 overflows to -567,053,209 when cast to int.

The connector log shows:
{noformat}
Unauthorized to execute splitVector command: not authorized on <db> to execute command { splitVector: "<db>.<collection>", ... }, fallback to SampleSplitter
Collection <db>.<collection> going to sample -567053209 records into 30150 chunks
{noformat}
Then MongoDB fails with:
{noformat}
Caused by: com.mongodb.MongoCommandException: Command failed with error 28747: 'size argument to $sample must not be negative'
{noformat}
h3. Expected behavior

SampleBucketSplitStrategy should calculate a positive sample size using long arithmetic.

For example, with:
{noformat}
numChunks = 30150
samplesPerChunk = 20
count = 3727914087
{noformat}
The expected sample size should be:
{noformat}
min(30150 * 20, 3727914087) = 603000
{noformat}
h3. Actual behavior

The sample size becomes negative because count is cast to int before Math.min:
{noformat}
Math.min(603000, (int) 3727914087)
= Math.min(603000, -567053209)
= -567053209
{noformat}
This negative value is passed to MongoDB $sample, and MongoDB rejects it.
h3. Reproduction conditions

This issue occurs when all of the following are true:
 * MongoDB CDC connector is used.
 * scan.incremental.snapshot.enabled is enabled.
 * The collection is split through SampleBucketSplitStrategy.
 * splitVector is unauthorized or fails, so the connector falls back to Sample Splitter.
 * The MongoDB collection document count is greater than Integer.MAX_VALUE.

h3. Suggested fix

Use long arithmetic when calculating the requested sample size, and only convert to int after validating the range.

For example:
{code:java}
long requestedSamples =
        Math.min((long) numChunks * splitContext.getSamplesPerChunk(), count);

checkArgument(
        requestedSamples <= Integer.MAX_VALUE,
        "The requested sample size exceeds Integer.MAX_VALUE: %s",
        requestedSamples);

int numberOfSamples = (int) requestedSamples;
{code}
It may also be safer to calculate chunkSizeInBytes using long arithmetic:
{code:java}
long chunkSizeInBytes = splitContext.getChunkSizeMB() * 1024L * 1024L;
{code}
instead of:
{code:java}
long chunkSizeInBytes = splitContext.getChunkSizeMB() * 1024 * 1024;
{code}



--
This message was sent by Atlassian Jira
(v8.20.10#820010)