fix: flush all records before closing the producer

[Cali0707]

Fixes a possible data race: we don't always flush all records before we stop receiving them.

Proposed Changes

• After we stop allowing records to be sent, we should call flush one more time on the producer

[knative-prow]

[APPROVALNOTIFIER] This PR is APPROVED

This pull-request has been approved by: Cali0707

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

The pull request process is described here

Needs approval from an approver in each of these files:

• OWNERS [Cali0707]

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

[Cali0707]

/cc @pierDipi

This shouldn't fix all the problems we are seeing, but this is a possible data race so I figured it's best to fix it

[codecov]

Codecov Report

All modified and coverable lines are covered by tests ✅

Project coverage is 49.72%. Comparing base (aacc057) to head (88f66e2).
Report is 9 commits behind head on main.

Additional details and impacted files

@@              Coverage Diff              @@
##               main    #3853       +/-   ##
=============================================
- Coverage     73.70%   49.72%   -23.98%     
=============================================
  Files           100      246      +146     
  Lines          3419    14819    +11400     
  Branches        292        0      -292     
=============================================
+ Hits           2520     7369     +4849     
- Misses          723     6695     +5972     
- Partials        176      755      +579     

Flag	Coverage Δ
java-unittests	?

Flags with carried forward coverage won't be shown. Click here to find out more.

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

[pierDipi]

/retest

[pierDipi]

We call flush before closing here

eventing-kafka-broker/data-plane/receiver/src/main/java/dev/knative/eventing/kafka/broker/receiver/impl/IngressProducerReconcilableStore.java

Lines 253 to 258 in 49ad678

	public Future<Void> close() {
	return producer.flush().compose(s -> closeNow(), c -> {
	logger.error("Failed to flush producer", c);
	return closeNow();
	});
	}

The idea of this class is to mimic the core Kafka producer interface and expose "flush" and "close" as separate methods

[Cali0707]

Ah that makes sense - I think our issue is that we need to flush after we stop receiving requests with events, but before we close the producer. Currently, we:

1. Flush
2. Stop receiving events
3. Close the producer
   This leads to a data race which causes a whole bunch of metadata requests (which aren't present now that we have this change)

I think we could fix this f we updated the producer to either:

1. Have another method on the interface to "mark the producer as closed" and then called markedClosed -> flush -> close, or
2. Have the flush method set the closed variable so that the producer stops receiving events

[pierDipi]

I think we could fix this f we updated the producer to either:

1. Have another method on the interface to "mark the producer as closed" and then called markedClosed -> flush -> close, or
2. Have the flush method set the closed variable so that the producer stops receiving events

Isn't this exactly what the returned future (in close) communicates to the caller ?

[pierDipi]

Meaning that with the changes in this PR we're doing 1 and 2 as we:

• set closed to true (line 125) -> stop accepting events (lines 69-70)
• flush
• close

[Cali0707]

Yeah, that is what is happening currently. I guess what I meant was to set closed to true in the flush method we have in the interface, or call flush again here

[Cali0707]

/retest-required

[creydr]

/retest-required

[creydr]

/lgtm

[Cali0707]

/retest-required

[Cali0707]

/hold

The warnings are still present in the newest set of logs, looks like this didn't fix it properly and earlier tests we were just lucky :(

[creydr]

/lgtm cancel

[Cali0707]

/cc @creydr @pierDipi

I think we still want to have this change, otherwise there is technically a case where we close the producer while there are still pending sends (since we wait for the eventQueue to be empty, but that only means we processed the event, not that the event's promise completed)

[Cali0707]

/retest-required

[knative-prow]

@Cali0707: The following tests failed, say /retest to rerun all failed tests or /retest-required to rerun all mandatory failed tests:

Test name	Commit	Details	Required	Rerun command
upgrade-tests_eventing-kafka-broker_main	88f66e2	link	true	/test upgrade-tests
reconciler-tests-keda_eventing-kafka-broker_main	88f66e2	link	true	/test reconciler-tests-keda

Your PR dashboard.

Instructions for interacting with me using PR comments are available here. If you have questions or suggestions related to my behavior, please file an issue against the kubernetes-sigs/prow repository. I understand the commands that are listed here.