[
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)
