Current state: Under Discussion
Discussion thread: here
JIRA: KAFKA-7996
Released: (Not decided yet)
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
As of 2.2.0, KafkaStream#close
works like the following:
StreamThread
s, GlobalStreamThread
, and AdminClient
. StreamThread
and GlobalStreamThread
have their own Producer
s and Consumer
s. All close methods are called without any timeout.KafkaStreams
state into NOT_RUNNING
. (see KafkaStreams#waitOnState
)However, for Producer
and AdminClient
's default close timeout is Duration.ofMillis(Long.MAX_VALUE)
, Kafka streams application may take a long time or even hang up for closing internal Producer
s or AdminClient
.
To resolve this problem, this KIP proposes to provide a close timeout parameter for closing internal clients.
There are 3 alternatives for providing close timeout parameter.
This approach makes any change on the public interface; it just uses a constant like KafkaStreams#DEFAULT_CLOSE_TIMEOUT
as a client close timeout.
Easy to implement.
Users have any way to configure close timeout. Moreover, some users may want current close timeout, i.e., Duration.ofMillis(Long.MAX_VALUE)
.
Adding a new configuration option, close.wait.ms
, and allow users to configure the close timeout. (default: 500ms)
Easy to configure.
Adds a new configuration option, which is already so many.
By adding a new optional parameter (closeTimeout
) to the constructor of KafkaStream
s, allow users to configure the close timeout if required (default: 500ms):
public KafkaStreams(final Topology topology, final Properties props, final Duration closeWaitTime); public KafkaStreams(final Topology topology, final Properties props, final KafkaClientSupplier clientSupplier, final Duration closeWaitTime); public KafkaStreams(final Topology topology, final Properties props, final Time time, final Duration closeWaitTime); public KafkaStreams(final Topology topology, final Properties props, final KafkaClientSupplier clientSupplier, final Time time, final Duration closeWaitTime); |
Allows the users to configure the close timeout, without adding new configuration option.
This approach adds 4 overload constructors; Since there are already 2 required parameters and 2 optional parameters, there are 4 constructors and now becomes 8 constructors - it is too many.
This problem can be mitigated by deprecating the overloaded constructors and providing KafkaStreams.Builder
instead. However, this issue is above the scope of this KIP.
(under discussion)
None.
By passing the existing unit tests.
This approach has an advantage in fixing the default close timeout inconsistency between Consumer
and Producer
, AdminClient
. However, this inconsistency was intended; The current `KafkaConsumer#close`'s default timeout, 30 seconds, was introduced in KIP-102 (0.10.2.0). To summarize, there are two differences between Consumer
and Producer
;
1. Consumer
s don't have large requests.
2. Consumer#close
is affected by consumer coordinator, whose close operation is affected by request.timeout.ms
.
By the above reasons, Consumer's default timeout was set a little bit different. So, we don't need to align the parity.
Moreover, this approach does not allow the users to configure a close timeout, and breaks backward compatibility. So rejected.