root
/
kafka
mirror of https://github.com/apache/kafka.git
1
0
Fork 0
Code Issues Actions 3 Packages Projects Releases Wiki Activity

addressed by comments

This commit is contained in:
m1a2st 2025-10-07 22:15:58 +08:00
parent c231f07798
commit 0b6f7ba0a9
2 changed files with 4 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
streams/src/main/java/org/apache/kafka/streams/CloseOptions.java
View File

@ -81,11 +81,7 @@ public class CloseOptions {
* @return this {@code CloseOptions} instance.
*/
public CloseOptions withTimeout(final Duration timeout) {
if (timeout == null) {
this.timeout = Optional.of(Duration.ofMillis(Long.MAX_VALUE));
} else {
this.timeout = Optional.of(timeout);
}
this.timeout = Optional.ofNullable(timeout);
return this;
}

5
streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java
View File

@ -1635,8 +1635,9 @@ public class KafkaStreams implements AutoCloseable {
public synchronized boolean close(final org.apache.kafka.streams.CloseOptions options) throws IllegalArgumentException {
Objects.requireNonNull(options, "options cannot be null");
final CloseOptionsInternal optionsInternal = new CloseOptionsInternal(options);
final String msgPrefix = prepareMillisCheckFailMsgPrefix(optionsInternal.timeout().get(), "timeout");
final long timeoutMs = validateMillisecondDuration(optionsInternal.timeout().get(), msgPrefix);
final Duration timeout = optionsInternal.timeout().orElse(Duration.ofMillis(Long.MAX_VALUE));
final String msgPrefix = prepareMillisCheckFailMsgPrefix(timeout, "timeout");
final long timeoutMs = validateMillisecondDuration(timeout, msgPrefix);
if (timeoutMs < 0) {
throw new IllegalArgumentException("Timeout can't be negative.");
}