...
- Launch a daemon thread that closes
StreamThread
s,GlobalStreamThread
, andAdminClient
.StreamThread
andGlobalStreamThread
have their ownProducer
s andConsumer
s. All close methods are called without any timeout. - Wait for the daemon thread to complete its job and change
KafkaStreams
state intoNOT_RUNNING
. (seeKafkaStreams#waitOnState
)
In short, the timeout
semantics in KafkaStreams#close
function is not for "enforcing to complete closing before this timeout", but "try to block on close for a maximum of this time, and if it still not complete then give up and return false", as described in the Javadoc:
Shutdown this {@code KafkaStreams} by signaling all the threads to stop, and then wait up to the timeout for the threads to join.
The problem is, the close semantics of internal clients (i.e., Producer
s, Consumer
s, and AdminClient
) are different from each other. Consumer
is closed with a default timeout of KafkaConsumer#DEFAULT_CLOSE_TIMEOUT_MS
(30 seconds). In contrast, However, for Producer
and AdminClient
's default close timeout is Duration.ofMillis(Long.MAX_VALUE)
. By this reason, 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 improve KafkaSteams#close
provide a close timeout parameter for closing internal clients.
Public Interfaces
There are 3 alternatives for providing close timeout parameter.
Type A. fix the close timeout as a constant
This approach makes This KIP make any change on the public interfacesinterface; however, it makes a subtle change to the existing API's semantics. If this KIP is accepted, documenting these semantics with as much detail as possible may much better.
Proposed Changes
...
it just uses a constant like KafkaStreams#DEFAULT_CLOSE_TIMEOUT
as a client close timeout.
Pros
Easy to implement.
Cons
Users have any way to configure close timeout. Moreover, some users may want current close timeout, i.e., Duration.ofMillis(Long.MAX_VALUE)
.
Type B.
...
Take the given timeout parameter into KafkaStreams#close
operation; that is,
It tries to close the resources in the remaining
timeout
as far as possible.If succeeded, simply return; if not, close remaining resources with default close timeout.
Provide a new configuration option
Adding a new configuration option, close.wait.ms
, and allow users to configure the close timeout. (default: 500ms)
Pros
Easy to configure.
Cons
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 KafkaStream
s, allow users to configure the close timeout if required (default: 500ms):
Code Block | ||
---|---|---|
| ||
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); |
Pros
Allows the users to configure the close timeout, without adding new configuration option.
Cons
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.
Proposed Changes
(under discussionWith this approach, the internal client hang-up problem can be resolved. Moreover, KafkaStreams#close
now respects users' intention without changing the semantics. (i.e., it *tries* to close the application in given timeout, at least.)
Compatibility, Deprecation, and Migration Plan
...
By passing the existing unit tests.
Rejected Alternatives
...
Change the default close timeout for [Producer, AdminClient]#close
This approach does not allow the users to configure a close timeout, and breaks backward compatibility. So rejected
These options have very limited functionality, so rejected.
Add default timeout to Producer
, AdminClient
only (i.e., timeout parameter is not taken account into KafkaStreams#close
operation)
This approach is the simplest one; however, from the user's perspective, this approach totally ignores the user's intention.