debezium/dbz#1081 add capturing filter strategy for single event processing

[vsantonastaso]

closes debezium/dbz#1081

Context

This PR introduces a pluggable CapturingFilterStrategy for the @Capturing annotation for user-defined filtering mechanism.
Users can implement custom filters that receive the full CapturingEvent<SourceRecord, SourceRecord> and decide whether to capture each event.

Changes

The @Capturing annotation now supports a filter attribute:

@Capturing(filter = MyFilter.class)
public void onEvent(CapturingEvent<SourceRecord, SourceRecord> event) { ... }

The filter attribute is mutually exclusive with destination and engine with a build time validaition.

A dedicated filter registry handles filter-based invokers separately from destination-based ones. Events are dispatched through a fallback chain: filter registry first, then object invoker with deserializer, then default registry (destination match + all destination fallback).

A built-in CapturingFieldsFilterStrategy abstract class is provided for field-based filtering:

@ApplicationScoped
public class MyFilter extends CapturingFieldsFilterStrategy {
    public MyFilter() {
        super(Set.of("name", "price"));
    }
}

This captures only events that impact the specified fields. In particular, for Create/Read event, checks if the field exists in the schema, For Update event checks if the field value changed, For Delete one checks if the field existed, Truncate/Message events are rejected.

Documentation

updated with debezium/debezium#7494

[kmos]

Why use an abstract class for field-based filtering? Wouldn’t it be possible to add a field directly to the @Capturing annotation which take a set?

[vsantonastaso]

Why use an abstract class for field-based filtering? Wouldn’t it be possible to add a field directly to the @Capturing annotation which take a set?

Good question.
My idea is to provide a generic interface, CapturingFilterStrategy, to support different types of filtering. It is not limited to field-based filtering (which would just require a set of strings), but can handle any filtering logic, even when different parameters are needed.
For this reason, the @capturing annotation only accepts an implementation of CapturingFilterStrategy.
For field-based filtering, I’ve provided an out-of-the-box implementatio, CapturingFieldsFilterStrategy, that is an abstract class to keep @capturing generic and avoid adding specific parameters, users need to extend this class and define the fields they want to filter.

[kmos]

CapturingFilterTest and CapturingFieldsFilterTest aren't available for NoSql database?

[vsantonastaso]

I missed it in NoSql test suite. Good catch

[vsantonastaso]

Just added logic into CapturingFieldsFilterStrategy for noSql connector (handle json payload in addition to Struct ones)