added custom broker filter

[astelmashenko]

A custom broker implementation consists of few components:

1. Feature/natsjs broker step1 #715 broker controller, to be able to specify NatsJetstreamBroker class for a broker CRD
2. added trigger controller and reconciler #716 trigger controller, to receive events for eventing triggers which are subscribed to the NatsJetstreamBroker class and reconcile subscriptions to JetStream
3. custom broker ingress #719 ingress, which accepts http requests to broker and stores it into JetStream stream storage
4. filter, which creates consumers to JetStream, get messages, filter them out and sends to targets according to triggers configuration
5. configuration is a set of yaml files describe CRD, service accounts, roles and permissions of above components
6. e2e tests/docs/examples

controller.go
Runs when the pod starts. Connects to NATS, then watches Kubernetes for Trigger resources. When someone creates, updates, or deletes a Trigger, it tells the reconciler.

reconciler.go
When a Trigger changes, the reconciler checks:

• Does the broker exist? Is it ready?
• Is it our broker type (NatsJetStreamBroker)?
• Where should events be delivered? (subscriber URL from trigger status)
• Is there a dead letter sink for failed events?

If everything checks out, it tells the ConsumerManager to start pulling events for that trigger.

consumer.go
Manages pull subscriptions to NATS JetStream. For each trigger, it creates a subscription and runs a fetch loop - a continuous loop that pulls
messages in batches (like checking your mailbox every few seconds). When messages arrive, it hands them to the handler.

handler.go
Does the actual work for each message:

1. Convert - turns the NATS message into a CloudEvent
2. Filter - checks if the event matches the trigger's filter rules (e.g., only events of type order.created). No match = skip
3. Deliver - sends the event to the subscriber via HTTP
4. Handle result:
   - Success → ACK (tell NATS "done, delete it")
   - Temporary failure → NACK (tell NATS "retry later")
   - Permanent failure → send to dead letter sink if configured, then TERM (give up)

A note on retries, handler implements retries based on Nats Jetstream consumer retries, so it does not do retires based on golang in-memory implementation, that's why there noRetry with request timeout only.

[knative-prow]

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: astelmashenko

The full list of commands accepted by this bot can be found here.

The pull request process is described here

Details

Needs approval from an approver in each of these files:

• OWNERS [astelmashenko]

Approvers can indicate their approval by writing /approve in a comment
Approvers can cancel approval by writing /approve cancel in a comment

[codecov]

Codecov Report

❌ Patch coverage is 37.31343% with 294 lines in your changes missing coverage. Please review.
✅ Project coverage is 32.54%. Comparing base (28230c5) to head (d7c5696).

Files with missing lines	Patch %	Lines
pkg/broker/filter/handler.go	30.86%	111 Missing and 1 partial ⚠️
pkg/broker/filter/consumer.go	39.45%	86 Missing and 3 partials ⚠️
pkg/broker/filter/controller.go	12.50%	77 Missing ⚠️
pkg/broker/filter/reconciler.go	86.15%	6 Missing and 3 partials ⚠️
cmd/filter/main.go	0.00%	7 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #720      +/-   ##
==========================================
+ Coverage   31.90%   32.54%   +0.64%     
==========================================
  Files          58       63       +5     
  Lines        3470     3939     +469     
==========================================
+ Hits         1107     1282     +175     
- Misses       2288     2575     +287     
- Partials       75       82       +7     

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[astelmashenko]

/assign @Cali0707