Current state: Under Discussion
Discussion thread: here
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:
- Launch a daemon thread that closes
GlobalStreamThreadhave their own
Consumers. All close methods are called without any timeout.
- Wait for the daemon thread to complete its job and change
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
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.
Type A. fix the close timeout as a constant
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.,
Type B. Provide a new configuration option
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.
Type C. Extend KafkaStreams constructor
By adding a new optional parameter (
closeTimeout) to the constructor of
KafkaStreams, 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.
Compatibility, Deprecation, and Migration Plan
By passing the existing unit tests.
Change the default close timeout for [Producer, AdminClient]#close
This approach has an advantage in fixing the default close timeout inconsistency between
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
Consumers don't have large requests.
Consumer#close is affected by consumer coordinator, whose close operation is affected by
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.