Filter DCB subscriptions server-side

[johanhaleby]

DCB subscriptions filtered every event in process. DcbSubscriptions subscribed with Filter.all() and threw away most events in the consumer thread, so a read model that cares about a few event types or one tag boundary still received the whole DCB stream. Under load that is a lot of wasted delivery.

This makes DCB filtering server-side.

• DcbSubscriptionFilter, wrapping a DcbQuery, is now a first-class SubscriptionFilter alongside the stream OccurrentSubscriptionFilter. One converter turns a DcbQuery into a single change stream $match: types to $in, excluded types to $nin, tags to $all on the indexed dcbTags array, items OR-ed, and dcbposition > 0 always so only DCB events arrive. The Spring and native MongoDB models both use it. The in-memory model matches the same query in process.
• DcbSubscriptions now subscribes with the filter and keeps its in-process check only as a correctness floor, since it wraps an arbitrary Subscribable and cannot assume the backend filters. A capable backend reduces the volume, the floor guarantees correctness everywhere.
• CatchupSubscriptionModel had one upfront guard that rejected every filter except the stream one. That guard belongs to the stream catch-up path, which converts the filter into an Occurrent Filter, so it now lives only there. The DCB path passes the filter straight to the inner model.
• The dcbTags field name was a private constant duplicated in the event store. It is now the public OccurrentCloudEventMongoDocumentMapper.DCB_TAGS_INDEX_FIELD, the one storage contract the writer and the subscription reader share.

On the open question of whether to split the Filter class itself: no. SubscriptionFilter is already the common seam, so the clean split is stream Filter for stream subscriptions and DcbQuery for DCB subscriptions, each an implementation over shared Mongo plumbing. There is no shared structure today that a common filter base would capture without inventing it. ADR 0023 records this.

Verification

• rtk mvn -pl subscription/mongodb/common/base -am -Dtest=DcbSubscriptionFilterConverterTest test: Tests run: 8, Failures: 0, Errors: 0.
• rtk mvn -pl subscription/inmemory -Dtest=InMemorySubscriptionModelDcbFilterTest test plus the existing in-memory suite: green.
• rtk mvn -pl framework/spring-boot-starter-mongodb -Dtest=DcbServerSideSubscriptionFilteringMongoTest test: Tests run: 4, Failures: 0, Errors: 0. This subscribes the model directly, not through DcbSubscriptions, so a non-matching delivered event would mean the server-side $match is wrong.
• Catch-up regression: CatchupSubscriptionModelTest, DcbCatchupSubscriptionModelTest, DcbCatchupSubscriptionModelMongoTest (23 total), plus the starter DcbCatchupSubscriptionAutoConfigurationMongoTest and OccurrentMongoAutoConfigurationCombinedModeTest: green.
• rtk mvn -pl dsl/dcb-dsl/blocking test: green, including the DCB DSL routing through the new filter.

[github-actions]

Qodana Community for JVM

76 new problems were found

Inspection name	Severity	Problems
Kotlin Maven Plugin misconfigured	🔶 Warning	29
Constant values	🔶 Warning	25
Nullability and data flow problems	🔶 Warning	14
'Optional' used as field or parameter type	🔶 Warning	2
Pointless boolean expression	🔶 Warning	2
'equals()' called on classes which don't override it	🔶 Warning	1
Result of method call ignored	🔶 Warning	1
Function or property has platform type	◽️ Notice	2

View the detailed Qodana report

To be able to view the detailed Qodana report, you can either:

• Register at Qodana Cloud and configure the action
• Use GitHub Code Scanning with Qodana
• Host Qodana report at GitHub Pages
• Inspect and use qodana.sarif.json (see the Qodana SARIF format for details)

To get *.log files or any other Qodana artifacts, run the action with upload-result option set to true,
so that the action will upload the files as the job artifacts:

      - name: 'Qodana Scan'
        uses: JetBrains/[email protected]
        with:
          upload-result: true

Contact Qodana team

Contact us at [email protected]

• Or via our issue tracker: https://jb.gg/qodana-issue
• Or share your feedback: https://jb.gg/qodana-discussions

[Copilot]

Pull request overview

This PR introduces server-side filtering for Dynamic Consistency Boundary (DCB) subscriptions by making DcbSubscriptionFilter (wrapping a DcbQuery) a first-class SubscriptionFilter, and translating it into a MongoDB change stream $match stage (plus equivalent in-memory matching). It reduces wasted delivery/consumer-side filtering by pushing DCB query semantics down to capable backends while retaining an in-process correctness floor where needed.

Changes:

• Add DcbSubscriptionFilter and a shared DcbQuery -> change stream $match converter, and route MongoDB (Spring + native) subscription models through it.
• Update the DCB DSL (DcbSubscriptions) to subscribe with DcbSubscriptionFilter while keeping post-filtering as a correctness floor.
• Remove the overly broad filter-type guard from the DCB catch-up path (keep it only for the stream catch-up path) and expose the dcbTags storage field constant as a shared contract.

Reviewed changes

Copilot reviewed 19 out of 19 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
subscription/util/blocking/catchup-subscription/src/main/java/org/occurrent/subscription/blocking/durable/catchup/CatchupSubscriptionModel.java	Moves the “only stream filter supported” guard into the stream path so DCB mode can accept DCB filters.
subscription/mongodb/spring/common/src/main/java/org/occurrent/subscription/mongodb/spring/internal/ApplyFilterToChangeStreamOptionsBuilder.java	Adds server-side change stream filtering for DcbSubscriptionFilter in Spring Mongo subscriptions.
subscription/mongodb/native/blocking/src/main/java/org/occurrent/subscription/mongodb/nativedriver/blocking/NativeMongoSubscriptionModel.java	Adds server-side change stream filtering for DcbSubscriptionFilter in the native Mongo driver subscription model.
subscription/mongodb/common/base/src/main/java/org/occurrent/subscription/mongodb/internal/DcbSubscriptionFilterConverter.java	Implements conversion from DcbQuery to a single change stream $match stage (incl. dcbposition > 0 guard).
subscription/mongodb/common/base/src/test/java/org/occurrent/subscription/mongodb/internal/DcbSubscriptionFilterConverterTest.java	Adds unit tests validating the $match stage structure produced for various DcbQuery shapes.
subscription/mongodb/common/base/pom.xml	Adds DCB API dependency and test dependencies needed for the new converter tests.
subscription/inmemory/src/main/java/org/occurrent/subscription/inmemory/InMemorySubscriptionModel.java	Extends in-memory subscriptions to honor DcbSubscriptionFilter by matching DCB queries in-process.
subscription/inmemory/src/main/java/org/occurrent/subscription/inmemory/InMemorySubscription.java	Switches from holding a stream Filter to a general Predicate<CloudEvent> matcher.
subscription/inmemory/src/test/java/org/occurrent/subscription/inmemory/InMemorySubscriptionModelDcbFilterTest.java	Adds in-memory tests ensuring DCB filter matching works (tags, types, and position guard).
subscription/inmemory/pom.xml	Adds DCB API dependency required for in-memory DCB filtering.
subscription/core/src/main/java/org/occurrent/subscription/DcbSubscriptionFilter.java	Introduces DcbSubscriptionFilter as a SubscriptionFilter implementation wrapping a DcbQuery.
subscription/core/pom.xml	Adds DCB API dependency to expose DcbSubscriptionFilter in subscription core.
framework/spring-boot-starter-mongodb/src/test/java/org/occurrent/springboot/mongo/blocking/DcbServerSideSubscriptionFilteringMongoTest.java	Adds end-to-end Spring Boot test proving server-side DCB change stream filtering works.
eventstore/mongodb/spring/blocking/src/main/java/org/occurrent/eventstore/mongodb/spring/blocking/SpringMongoEventStore.java	Removes duplicated dcbTags field constant and uses the mapper’s shared constant.
eventstore/mongodb/common/src/main/java/org/occurrent/eventstore/mongodb/internal/OccurrentCloudEventMongoDocumentMapper.java	Promotes DCB_TAGS_INDEX_FIELD to a public constant as a storage contract shared by writer and subscriptions.
dsl/dcb-dsl/blocking/src/main/kotlin/org/occurrent/dsl/dcb/blocking/DcbSubscriptions.kt	Updates Kotlin DCB DSL to subscribe using DcbSubscriptionFilter instead of subscribing with Filter.all().
dsl/dcb-dsl/blocking/src/main/java/org/occurrent/dsl/dcb/blocking/DcbSubscriptions.java	Updates Java DCB DSL to subscribe using DcbSubscriptionFilter instead of subscribing with Filter.all().
doc/architecture/decisions/0023-server-side-filtering-for-dcb-subscriptions.md	Adds ADR documenting the design decision and tradeoffs for server-side DCB subscription filtering.
changelog.md	Documents the new server-side filtering behavior and the new public constant contract.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[johanhaleby]

Landed via the consolidated #234.