[KAFKA-13870] support both Suppressed untilTimeLimit and maxBytes without using emitEarlyWhenFull()

• Type: New Feature
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Labels: needs-kip
• Assignee: Unassigned
• Reporter: Anil
• Created: 04/May/22 05:53
• Updated: 04/May/22 16:47

Description

My use case is to use  ** untilTimeLimit with maxBytes, but when the buffer is full, the application is breaking, but with using emitEarlyWhenFull application is not breaking but{{}} it sends out the same key record multiple times in a particular window when the buffer exceeds max bytes 

for eg:-
Suppressed.untilTimeLimit(Duration.ofMinutes(15),Suppressed.BufferConfig.maxBytes(10000).emitEarlyWhenFull())
 
messages flow : (A,1) (A,2) (A,3) -> aggregation result : (A,6) . suppose here, the buffer is full, (A,6) will be sent downstream. Let's suppose (A,4) comes now in the same tumbling window.
 
current response:- the aggregation will continue and eventually (A,10) will be emitted
 
but our application expected (A,4),  so the request for the feature is that window should be happening with window time(untilTimeLimit) or Buffer(maxByte) should full, in either of these two conditions met, a new window should be created and data should be emitted