Copilot commented on code in PR #72: URL: https://github.com/apache/flink-connector-mongodb/pull/72#discussion_r4501779859
########## AGENTS.md: ########## @@ -0,0 +1,217 @@ +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# Flink MongoDB Connector AI Agent Instructions + +This file provides guidance for AI coding agents working with the Apache Flink MongoDB connector codebase. + +## Prerequisites + +- Java 11, 17, or 21 (the `main` branch build matrix; release branch `v2.0` also supports Java 8) +- On Java 11, add `-Pjava11-target` to every build and test command below. `flink-connector-parent` defaults `maven.compiler.target` to 17, which javac 11 rejects; CI adds the profile on its Java 11 jobs. +- Maven 3.8.6 (there is no Maven wrapper in this repository; use `mvn` directly) +- Git +- Docker (every `*ITCase` starts MongoDB through Testcontainers, including sharded-cluster tests via `MongoShardedContainers`) +- Unix-like environment (Linux, macOS) +- The connector builds against `flink.version` in the root `pom.xml`. A change must also work against every Flink version in `.github/workflows/push_pr.yml` and the `weekly.yml` matrix. +- MongoDB server compatibility is tested via Maven profiles (`mongodb4`, `mongodb5`, `mongodb6`, `mongodb7`, each pinning `mongodb.version` to a specific server release); CI runs the full test suite once per profile. + +## Commands + +### Build + +- Build without tests: `mvn clean package -DskipTests` +- Full build with tests: `mvn clean verify` +- Build against a specific MongoDB server version (what CI does): `mvn clean verify -Pmongodb7` Review Comment: These commands omit the end-to-end tests: the e2e module disables normal test executions and enables them only with `run-end-to-end-tests`. The shared CI workflow enables that profile and supplies `distDir`. Please distinguish plain `verify` from full-suite validation and document those flags so agents do not report full validation without testing against a Flink distribution. ########## AGENTS.md: ########## @@ -0,0 +1,217 @@ +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# Flink MongoDB Connector AI Agent Instructions + +This file provides guidance for AI coding agents working with the Apache Flink MongoDB connector codebase. + +## Prerequisites + +- Java 11, 17, or 21 (the `main` branch build matrix; release branch `v2.0` also supports Java 8) +- On Java 11, add `-Pjava11-target` to every build and test command below. `flink-connector-parent` defaults `maven.compiler.target` to 17, which javac 11 rejects; CI adds the profile on its Java 11 jobs. +- Maven 3.8.6 (there is no Maven wrapper in this repository; use `mvn` directly) +- Git +- Docker (every `*ITCase` starts MongoDB through Testcontainers, including sharded-cluster tests via `MongoShardedContainers`) +- Unix-like environment (Linux, macOS) +- The connector builds against `flink.version` in the root `pom.xml`. A change must also work against every Flink version in `.github/workflows/push_pr.yml` and the `weekly.yml` matrix. Review Comment: The weekly matrix includes Flink 1.19.2 and 1.20.1 jobs that explicitly check out `v2.0`, not `main`. Requiring a change on `main` to work against every weekly entry therefore assigns it release-branch compatibility targets that its CI does not test. Please scope this requirement to the entries that build the branch being changed. ########## AGENTS.md: ########## @@ -0,0 +1,217 @@ +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# Flink MongoDB Connector AI Agent Instructions + +This file provides guidance for AI coding agents working with the Apache Flink MongoDB connector codebase. + +## Prerequisites + +- Java 11, 17, or 21 (the `main` branch build matrix; release branch `v2.0` also supports Java 8) +- On Java 11, add `-Pjava11-target` to every build and test command below. `flink-connector-parent` defaults `maven.compiler.target` to 17, which javac 11 rejects; CI adds the profile on its Java 11 jobs. +- Maven 3.8.6 (there is no Maven wrapper in this repository; use `mvn` directly) +- Git +- Docker (every `*ITCase` starts MongoDB through Testcontainers, including sharded-cluster tests via `MongoShardedContainers`) +- Unix-like environment (Linux, macOS) +- The connector builds against `flink.version` in the root `pom.xml`. A change must also work against every Flink version in `.github/workflows/push_pr.yml` and the `weekly.yml` matrix. +- MongoDB server compatibility is tested via Maven profiles (`mongodb4`, `mongodb5`, `mongodb6`, `mongodb7`, each pinning `mongodb.version` to a specific server release); CI runs the full test suite once per profile. + +## Commands + +### Build + +- Build without tests: `mvn clean package -DskipTests` +- Full build with tests: `mvn clean verify` +- Build against a specific MongoDB server version (what CI does): `mvn clean verify -Pmongodb7` +- Build against another Flink version: `mvn clean verify -Dflink.version=<version>` +- Single module: `mvn clean install -DskipTests -pl flink-connector-mongodb` +- Skip non-essential checks while iterating: `-Dspotless.check.skip=true -Dcheckstyle.skip=true -Drat.skip=true` +- Dependency convergence check: `mvn clean install -DskipTests -Pcheck-convergence -Dflink.convergence.phase=install`. Off in plain local builds (`flink-connector-parent` binds `dependency-convergence` to `<phase>none</phase>` "as it interacts badly with shade-plugin"; the `check-convergence` profile re-binds it). PR/push CI (`push_pr.yml`) runs it via the shared-utils default `run_dependency_convergence: true`; weekly CI (`weekly.yml`) sets it to `false`. + +### Testing + +- `*Test` classes run in the `test` phase, `*ITCase` classes run in the `integration-test` phase. +- Single unit test class: `mvn test -pl flink-connector-mongodb -Dtest=MongoSampleSplitterTest` +- Single ITCase: `mvn verify -pl flink-connector-mongodb -Dtest=MongoSinkITCase` +- ArchUnit rules: `mvn test -pl flink-connector-mongodb -Dtest='*ArchitectureTest'`, then check `git status flink-connector-mongodb/archunit-violations/` (see Testing Standards) +- MongoDB in tests comes from `MongoTestUtil` (`createMongoDBContainer()`, `createMongoDBShardedContainers()`), which reads the target server version from the `mongodb.version` system property — set by whichever `mongodbN` profile is active, not hardcoded in test code. +- CI is defined by `apache/flink-connector-shared-utils` (`.github/workflows/ci.yml@ci_utils`); its Maven command line is the reference when a local run differs from CI. + +### Code Quality + +- Format code: `mvn spotless:apply` (skipped automatically on JDK 21 — google-java-format does not run there; format on 11 or 17) +- Check formatting: `mvn spotless:check` +- Checkstyle: `mvn checkstyle:check` (config: `tools/maven/checkstyle.xml`, suppressions: `tools/maven/suppressions.xml`) +- License headers: `mvn apache-rat:check` +- `japicmp` compares against `japicmp.referenceVersion` from the root `pom.xml` (`<connector version>-<flink version>` format) and only checks `@Public`/`@PublicEvolving` API. + +### Documentation + +- `docs/content/docs/connectors/` and `docs/content.zh/docs/connectors/` hold the DataStream and Table docs (English and Chinese, same file set). Option tables are hand-written HTML rows — nothing is generated from `ConfigOption` definitions. +- `docs/data/mongodb.yml` only declares the connector's release `version`/`variants` metadata consumed by the Flink docs site build; it is not a source of table option content. + +## Repository Structure + +### Modules + +- `flink-connector-mongodb` — The connector: `MongoSource` (FLIP-27, `Boundedness.BOUNDED` only — this is a batch/bounded source, there is no CDC or unbounded streaming mode), `MongoSink` (Sink V2), and the Table/SQL factory (`MongoDynamicTableFactory`). +- `flink-sql-connector-mongodb` — Shaded SQL jar; relocates `com.mongodb`/`org.bson`. +- `flink-connector-mongodb-e2e-tests` — End-to-end tests against a real Flink distribution. +- `flink-connector-mongodb-table-tests` — Table/SQL-layer integration tests kept in their own module. + +There is no `flink-python` module in this repository. + +### Key packages in `flink-connector-mongodb/src/main/java` + +- `org.apache.flink.connector.mongodb.source` — `@PublicEvolving`: `MongoSource`, `MongoSourceBuilder`. `source.enumerator` / `source.reader` / `source.split` are `@Internal`. + - `source.enumerator.splitter` implements the partitioning strategies (`PartitionStrategy`, `@PublicEvolving`): `DEFAULT` (the default for both `MongoSourceBuilder` and `scan.partition.strategy`: `SHARDED` for sharded collections, otherwise `SPLIT_VECTOR`), `SINGLE` (no split), `SAMPLE` (random-sample based), `SPLIT_VECTOR` (uses the `splitVector` command, standalone/replica-set only), `SHARDED` (reads chunk boundaries from `config.chunks`), `PAGINATION` (skip/limit based, works everywhere including Atlas). `MongoSplitters` dispatches on the strategy; every strategy except `DEFAULT` has its own `MongoXxxSplitter`. +- `org.apache.flink.connector.mongodb.sink` — `@PublicEvolving`: `MongoSink`, `MongoSinkBuilder`, `MongoSerializationSchema`. `sink.writer` (`MongoWriter`, contexts) is `@Internal`. +- `org.apache.flink.connector.mongodb.table` — Table/SQL layer: `MongoDynamicTableFactory` is `@Internal`; `FilterHandlingPolicy` is `@PublicEvolving`. Converters (`BsonToRowDataConverters`, `RowDataToBsonConverters`) and the lookup function (`MongoRowDataLookupFunction`) live here too. +- `org.apache.flink.connector.mongodb.common` — Shared, `@Internal` utilities: `MongoConnectionOptions`, and in `common.utils`: `MongoConstants`, `MongoSerdeUtils`, `MongoUtils`, `MongoValidationUtils`. Check these before writing new helpers. Review Comment: `MongoConnectionOptions` and its builder are both annotated `@PublicEvolving` in `common/config/MongoConnectionOptions.java`, not `@Internal`. Classifying them as internal gives agents incorrect API-compatibility guidance when changing connection configuration. Please distinguish the public-evolving configuration API from the internal utility classes. ########## AGENTS.md: ########## @@ -0,0 +1,217 @@ +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# Flink MongoDB Connector AI Agent Instructions + +This file provides guidance for AI coding agents working with the Apache Flink MongoDB connector codebase. + +## Prerequisites + +- Java 11, 17, or 21 (the `main` branch build matrix; release branch `v2.0` also supports Java 8) +- On Java 11, add `-Pjava11-target` to every build and test command below. `flink-connector-parent` defaults `maven.compiler.target` to 17, which javac 11 rejects; CI adds the profile on its Java 11 jobs. +- Maven 3.8.6 (there is no Maven wrapper in this repository; use `mvn` directly) +- Git +- Docker (every `*ITCase` starts MongoDB through Testcontainers, including sharded-cluster tests via `MongoShardedContainers`) +- Unix-like environment (Linux, macOS) +- The connector builds against `flink.version` in the root `pom.xml`. A change must also work against every Flink version in `.github/workflows/push_pr.yml` and the `weekly.yml` matrix. +- MongoDB server compatibility is tested via Maven profiles (`mongodb4`, `mongodb5`, `mongodb6`, `mongodb7`, each pinning `mongodb.version` to a specific server release); CI runs the full test suite once per profile. + +## Commands + +### Build + +- Build without tests: `mvn clean package -DskipTests` +- Full build with tests: `mvn clean verify` +- Build against a specific MongoDB server version (what CI does): `mvn clean verify -Pmongodb7` +- Build against another Flink version: `mvn clean verify -Dflink.version=<version>` +- Single module: `mvn clean install -DskipTests -pl flink-connector-mongodb` +- Skip non-essential checks while iterating: `-Dspotless.check.skip=true -Dcheckstyle.skip=true -Drat.skip=true` +- Dependency convergence check: `mvn clean install -DskipTests -Pcheck-convergence -Dflink.convergence.phase=install`. Off in plain local builds (`flink-connector-parent` binds `dependency-convergence` to `<phase>none</phase>` "as it interacts badly with shade-plugin"; the `check-convergence` profile re-binds it). PR/push CI (`push_pr.yml`) runs it via the shared-utils default `run_dependency_convergence: true`; weekly CI (`weekly.yml`) sets it to `false`. + +### Testing + +- `*Test` classes run in the `test` phase, `*ITCase` classes run in the `integration-test` phase. +- Single unit test class: `mvn test -pl flink-connector-mongodb -Dtest=MongoSampleSplitterTest` +- Single ITCase: `mvn verify -pl flink-connector-mongodb -Dtest=MongoSinkITCase` +- ArchUnit rules: `mvn test -pl flink-connector-mongodb -Dtest='*ArchitectureTest'`, then check `git status flink-connector-mongodb/archunit-violations/` (see Testing Standards) +- MongoDB in tests comes from `MongoTestUtil` (`createMongoDBContainer()`, `createMongoDBShardedContainers()`), which reads the target server version from the `mongodb.version` system property — set by whichever `mongodbN` profile is active, not hardcoded in test code. +- CI is defined by `apache/flink-connector-shared-utils` (`.github/workflows/ci.yml@ci_utils`); its Maven command line is the reference when a local run differs from CI. + +### Code Quality + +- Format code: `mvn spotless:apply` (skipped automatically on JDK 21 — google-java-format does not run there; format on 11 or 17) +- Check formatting: `mvn spotless:check` +- Checkstyle: `mvn checkstyle:check` (config: `tools/maven/checkstyle.xml`, suppressions: `tools/maven/suppressions.xml`) +- License headers: `mvn apache-rat:check` +- `japicmp` compares against `japicmp.referenceVersion` from the root `pom.xml` (`<connector version>-<flink version>` format) and only checks `@Public`/`@PublicEvolving` API. + +### Documentation + +- `docs/content/docs/connectors/` and `docs/content.zh/docs/connectors/` hold the DataStream and Table docs (English and Chinese, same file set). Option tables are hand-written HTML rows — nothing is generated from `ConfigOption` definitions. +- `docs/data/mongodb.yml` only declares the connector's release `version`/`variants` metadata consumed by the Flink docs site build; it is not a source of table option content. + +## Repository Structure + +### Modules + +- `flink-connector-mongodb` — The connector: `MongoSource` (FLIP-27, `Boundedness.BOUNDED` only — this is a batch/bounded source, there is no CDC or unbounded streaming mode), `MongoSink` (Sink V2), and the Table/SQL factory (`MongoDynamicTableFactory`). +- `flink-sql-connector-mongodb` — Shaded SQL jar; relocates `com.mongodb`/`org.bson`. +- `flink-connector-mongodb-e2e-tests` — End-to-end tests against a real Flink distribution. +- `flink-connector-mongodb-table-tests` — Table/SQL-layer integration tests kept in their own module. + +There is no `flink-python` module in this repository. + +### Key packages in `flink-connector-mongodb/src/main/java` + +- `org.apache.flink.connector.mongodb.source` — `@PublicEvolving`: `MongoSource`, `MongoSourceBuilder`. `source.enumerator` / `source.reader` / `source.split` are `@Internal`. + - `source.enumerator.splitter` implements the partitioning strategies (`PartitionStrategy`, `@PublicEvolving`): `DEFAULT` (the default for both `MongoSourceBuilder` and `scan.partition.strategy`: `SHARDED` for sharded collections, otherwise `SPLIT_VECTOR`), `SINGLE` (no split), `SAMPLE` (random-sample based), `SPLIT_VECTOR` (uses the `splitVector` command, standalone/replica-set only), `SHARDED` (reads chunk boundaries from `config.chunks`), `PAGINATION` (skip/limit based, works everywhere including Atlas). `MongoSplitters` dispatches on the strategy; every strategy except `DEFAULT` has its own `MongoXxxSplitter`. +- `org.apache.flink.connector.mongodb.sink` — `@PublicEvolving`: `MongoSink`, `MongoSinkBuilder`, `MongoSerializationSchema`. `sink.writer` (`MongoWriter`, contexts) is `@Internal`. +- `org.apache.flink.connector.mongodb.table` — Table/SQL layer: `MongoDynamicTableFactory` is `@Internal`; `FilterHandlingPolicy` is `@PublicEvolving`. Converters (`BsonToRowDataConverters`, `RowDataToBsonConverters`) and the lookup function (`MongoRowDataLookupFunction`) live here too. +- `org.apache.flink.connector.mongodb.common` — Shared, `@Internal` utilities: `MongoConnectionOptions`, and in `common.utils`: `MongoConstants`, `MongoSerdeUtils`, `MongoUtils`, `MongoValidationUtils`. Check these before writing new helpers. + +## Architecture Boundaries + +1. **Source (FLIP-27, bounded only).** `MongoSourceEnumerator` implements `SplitEnumerator<MongoSourceSplit, MongoSourceEnumState>` and assigns splits produced up front by a `MongoSplitAssigner`/`MongoSplitters` (splitting happens once at enumerator startup via the configured `PartitionStrategy`, not incrementally). `MongoSourceReader` / `MongoScanSourceSplitReader` read one split (a query range) to completion. There is no offset-commit or incremental-discovery machinery like Kafka's — a split is either fully read or not read at all. Review Comment: The source checkpoints partially read splits and restores their document-count offsets with `skip(offset)`, so recovery is not all-or-nothing. Splits are also generated lazily in `MongoScanSplitAssigner.getNext()`, not at enumerator startup. Please describe both behaviors accurately so agents preserve partial-progress recovery when changing the source. ########## AGENTS.md: ########## @@ -0,0 +1,217 @@ +<!-- +Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. +--> + +# Flink MongoDB Connector AI Agent Instructions + +This file provides guidance for AI coding agents working with the Apache Flink MongoDB connector codebase. + +## Prerequisites + +- Java 11, 17, or 21 (the `main` branch build matrix; release branch `v2.0` also supports Java 8) +- On Java 11, add `-Pjava11-target` to every build and test command below. `flink-connector-parent` defaults `maven.compiler.target` to 17, which javac 11 rejects; CI adds the profile on its Java 11 jobs. +- Maven 3.8.6 (there is no Maven wrapper in this repository; use `mvn` directly) +- Git +- Docker (every `*ITCase` starts MongoDB through Testcontainers, including sharded-cluster tests via `MongoShardedContainers`) +- Unix-like environment (Linux, macOS) +- The connector builds against `flink.version` in the root `pom.xml`. A change must also work against every Flink version in `.github/workflows/push_pr.yml` and the `weekly.yml` matrix. +- MongoDB server compatibility is tested via Maven profiles (`mongodb4`, `mongodb5`, `mongodb6`, `mongodb7`, each pinning `mongodb.version` to a specific server release); CI runs the full test suite once per profile. + +## Commands + +### Build + +- Build without tests: `mvn clean package -DskipTests` +- Full build with tests: `mvn clean verify` +- Build against a specific MongoDB server version (what CI does): `mvn clean verify -Pmongodb7` +- Build against another Flink version: `mvn clean verify -Dflink.version=<version>` +- Single module: `mvn clean install -DskipTests -pl flink-connector-mongodb` +- Skip non-essential checks while iterating: `-Dspotless.check.skip=true -Dcheckstyle.skip=true -Drat.skip=true` +- Dependency convergence check: `mvn clean install -DskipTests -Pcheck-convergence -Dflink.convergence.phase=install`. Off in plain local builds (`flink-connector-parent` binds `dependency-convergence` to `<phase>none</phase>` "as it interacts badly with shade-plugin"; the `check-convergence` profile re-binds it). PR/push CI (`push_pr.yml`) runs it via the shared-utils default `run_dependency_convergence: true`; weekly CI (`weekly.yml`) sets it to `false`. + +### Testing + +- `*Test` classes run in the `test` phase, `*ITCase` classes run in the `integration-test` phase. +- Single unit test class: `mvn test -pl flink-connector-mongodb -Dtest=MongoSampleSplitterTest` +- Single ITCase: `mvn verify -pl flink-connector-mongodb -Dtest=MongoSinkITCase` +- ArchUnit rules: `mvn test -pl flink-connector-mongodb -Dtest='*ArchitectureTest'`, then check `git status flink-connector-mongodb/archunit-violations/` (see Testing Standards) +- MongoDB in tests comes from `MongoTestUtil` (`createMongoDBContainer()`, `createMongoDBShardedContainers()`), which reads the target server version from the `mongodb.version` system property — set by whichever `mongodbN` profile is active, not hardcoded in test code. +- CI is defined by `apache/flink-connector-shared-utils` (`.github/workflows/ci.yml@ci_utils`); its Maven command line is the reference when a local run differs from CI. + +### Code Quality + +- Format code: `mvn spotless:apply` (skipped automatically on JDK 21 — google-java-format does not run there; format on 11 or 17) +- Check formatting: `mvn spotless:check` +- Checkstyle: `mvn checkstyle:check` (config: `tools/maven/checkstyle.xml`, suppressions: `tools/maven/suppressions.xml`) +- License headers: `mvn apache-rat:check` +- `japicmp` compares against `japicmp.referenceVersion` from the root `pom.xml` (`<connector version>-<flink version>` format) and only checks `@Public`/`@PublicEvolving` API. + +### Documentation + +- `docs/content/docs/connectors/` and `docs/content.zh/docs/connectors/` hold the DataStream and Table docs (English and Chinese, same file set). Option tables are hand-written HTML rows — nothing is generated from `ConfigOption` definitions. +- `docs/data/mongodb.yml` only declares the connector's release `version`/`variants` metadata consumed by the Flink docs site build; it is not a source of table option content. + +## Repository Structure + +### Modules + +- `flink-connector-mongodb` — The connector: `MongoSource` (FLIP-27, `Boundedness.BOUNDED` only — this is a batch/bounded source, there is no CDC or unbounded streaming mode), `MongoSink` (Sink V2), and the Table/SQL factory (`MongoDynamicTableFactory`). +- `flink-sql-connector-mongodb` — Shaded SQL jar; relocates `com.mongodb`/`org.bson`. +- `flink-connector-mongodb-e2e-tests` — End-to-end tests against a real Flink distribution. +- `flink-connector-mongodb-table-tests` — Table/SQL-layer integration tests kept in their own module. + +There is no `flink-python` module in this repository. + +### Key packages in `flink-connector-mongodb/src/main/java` + +- `org.apache.flink.connector.mongodb.source` — `@PublicEvolving`: `MongoSource`, `MongoSourceBuilder`. `source.enumerator` / `source.reader` / `source.split` are `@Internal`. + - `source.enumerator.splitter` implements the partitioning strategies (`PartitionStrategy`, `@PublicEvolving`): `DEFAULT` (the default for both `MongoSourceBuilder` and `scan.partition.strategy`: `SHARDED` for sharded collections, otherwise `SPLIT_VECTOR`), `SINGLE` (no split), `SAMPLE` (random-sample based), `SPLIT_VECTOR` (uses the `splitVector` command, standalone/replica-set only), `SHARDED` (reads chunk boundaries from `config.chunks`), `PAGINATION` (skip/limit based, works everywhere including Atlas). `MongoSplitters` dispatches on the strategy; every strategy except `DEFAULT` has its own `MongoXxxSplitter`. +- `org.apache.flink.connector.mongodb.sink` — `@PublicEvolving`: `MongoSink`, `MongoSinkBuilder`, `MongoSerializationSchema`. `sink.writer` (`MongoWriter`, contexts) is `@Internal`. +- `org.apache.flink.connector.mongodb.table` — Table/SQL layer: `MongoDynamicTableFactory` is `@Internal`; `FilterHandlingPolicy` is `@PublicEvolving`. Converters (`BsonToRowDataConverters`, `RowDataToBsonConverters`) and the lookup function (`MongoRowDataLookupFunction`) live here too. +- `org.apache.flink.connector.mongodb.common` — Shared, `@Internal` utilities: `MongoConnectionOptions`, and in `common.utils`: `MongoConstants`, `MongoSerdeUtils`, `MongoUtils`, `MongoValidationUtils`. Check these before writing new helpers. + +## Architecture Boundaries + +1. **Source (FLIP-27, bounded only).** `MongoSourceEnumerator` implements `SplitEnumerator<MongoSourceSplit, MongoSourceEnumState>` and assigns splits produced up front by a `MongoSplitAssigner`/`MongoSplitters` (splitting happens once at enumerator startup via the configured `PartitionStrategy`, not incrementally). `MongoSourceReader` / `MongoScanSourceSplitReader` read one split (a query range) to completion. There is no offset-commit or incremental-discovery machinery like Kafka's — a split is either fully read or not read at all. +2. **Sink (Sink V2, at-least-once only — no exactly-once/transactions).** `MongoWriter` batches `WriteModel<BsonDocument>` objects from `MongoSerializationSchema` and flushes via `MongoCollection.bulkWrite`. A flush is triggered by whichever comes first: `batchSize`, `batchIntervalMs` (via a background `ScheduledExecutorService`), or a checkpoint (`flush(boolean endOfInput)`, gated by `flushOnCheckpoint`). `write()` blocks on `mailboxExecutor.yield()` while `checkpointInProgress` is true, so no new records are batched mid-flush. A failed flush is stored in a `volatile Exception flushException` and rethrown from the next `write()`/`flush()` call — it is never silently dropped. +3. **Checkpointed state.** `MongoSourceSplitSerializer` (`CURRENT_VERSION = 0`) and `MongoSourceEnumStateSerializer` version the source's split/enumerator state. The sink is stateless (no writer-state serializer) since it has no transactions to recover. +4. **Table layer.** `MongoDynamicTableSource`/`MongoDynamicTableSink` wrap the DataStream connectors; `MongoConnectorOptions` holds every `ConfigOption`. `MongoFilterPushDownVisitor` and `FilterHandlingPolicy` control filter pushdown into the MongoDB query; `MongoShardKeysExtractor`/`MongoPrimaryKeyExtractor` handle upsert/shard-key semantics. +5. **Connector vs Flink.** Production code may depend only on `@Public`/`@PublicEvolving` Flink API outside connector and util packages (ArchUnit rule, see `ProductionCodeArchitectureTest`). Every Flink API used must exist with the same annotation in `flink.version`, because the connector is released for several Flink minor versions. + +## Common Change Patterns + +### Adding a Table/SQL option + +1. Define the `ConfigOption<T>` in `MongoConnectorOptions` +2. Register and validate it in `MongoDynamicTableFactory` / `MongoConfiguration` +3. Add a factory test and, when behaviour changes, an ITCase in `flink-connector-mongodb-table-tests` +4. Add the option row to `docs/content/docs/connectors/table/mongodb.md` and the same file under `docs/content.zh/` +5. Fill in the Release Notes field on the JIRA ticket + +### Adding a DataStream builder option + +1. Add it to `MongoSourceBuilder` or `MongoSinkBuilder`, with validation in `build()` +2. Add a builder unit test and, when behaviour changes, an ITCase +3. Document it in `docs/content/docs/connectors/datastream/mongodb.md` and the `.zh` copy + +### Changing checkpointed state + +1. Bump `MongoSourceSplitSerializer.CURRENT_VERSION` or the enumerator state serializer version, and keep a read path for every older version Review Comment: These serializers currently share one version: `MongoSourceEnumStateSerializer.getVersion()` returns `MongoSourceSplitSerializer.CURRENT_VERSION`, and its deserializer passes that version to the embedded split deserializer. Bumping an independent enumerator version would make those embedded splits fail with `Unknown version` unless that coupling is also changed. Please document the shared version and require compatible read paths in both serializers when either state format changes. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
