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

["Qiu Yanjun (Jira)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-40257?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18100946#comment-18100946 ] 

Qiu Yanjun commented on FLINK-40257:
------------------------------------

Thanks for reporting this. I checked the current apache/flink-cdc master code and independently reproduced the arithmetic with the values from the issue. The current expression returns -567053209, while the same calculation performed in long arithmetic returns the expected 603000.

The root cause is the narrowing cast in Math.min(numChunks * samplesPerChunk, (int) count). When count is greater than Integer.MAX_VALUE, the cast wraps to a negative int before Math.min is evaluated. Because the splitVector failure path falls back to SampleBucketSplitStrategy, this negative result reaches Aggregates.sample(int) and MongoDB rejects the $sample stage. There is a second overflow risk in numChunks * samplesPerChunk because that multiplication is also currently evaluated as int.

My proposed fix is to calculate the requested sample count entirely with long arithmetic, for example Math.min((long) numChunks * samplesPerChunk, count), validate that the result is positive and no greater than Integer.MAX_VALUE, and only then cast it to int for the MongoDB driver. I would also change chunkSizeInBytes to use long multiplication (chunkSizeMB * 1024L * 1024L) to avoid the adjacent int-overflow risk.

For regression coverage, I plan to extract the sample-count calculation into a package-visible helper and test at least: the reported count 3727914087 with 30150 chunks and 20 samples per chunk (expected 603000), the small-collection/full-sampling path, normal large collections, and an out-of-range requested sample count that should fail with a clear validation error rather than wrap.

[~soyeong], could you please assign FLINK-40257 to me? I would like to prepare the fix and regression tests.

> 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 3.6.0 (MongoDB)
> Flink 1.20.4
>            Reporter: soyeong choe
>            Priority: Major
>
> 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)