You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "mjsax (via GitHub)" <gi...@apache.org> on 2025/04/24 02:39:57 UTC
[PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
mjsax opened a new pull request, #19546:
URL: https://github.com/apache/kafka/pull/19546
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2059377283
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1776,10 +1777,13 @@ public void close() {
}
/**
- * Tries to close the consumer cleanly within the specified timeout. This method waits up to
- * {@code timeout} for the consumer to complete pending commits and leave the group.
+ * This method has been deprecated since Kafka 4.0 and should use {@link KafkaConsumer#close(CloseOptions)} instead.
+ * <p>
+ * Close the consumer with {@link CloseOptions.GroupMembershipOperation#DEFAULT default membership operation}
+ * cleanly within the specified timeout. This method waits up to
+ * {@code timeout} for the consumer to complete pending commits and may leave the group (if applicable).
Review Comment:
agreed, "if applicable" is pretty vague and I'd be left wondering if I saw this as a user
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2061074867
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated(since = "4.1")
@Override
- @SuppressWarnings("deprecation")
public void close(Duration timeout) {
delegate.close(timeout);
}
/**
- * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.
- * The thread which is blocking in an operation will throw {@link org.apache.kafka.common.errors.WakeupException}.
- * If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
+ * Close the consumer cleanly. {@link CloseOptions} allows to specify a timeout and a
+ * {@link CloseOptions.GroupMembershipOperation leave group behavior}.
+ * If no timeout is specified, the default timeout of 30 seconds is used.
+ * If no leave group behavior is specified, the {@link CloseOptions.GroupMembershipOperation#DEFAULT default
+ * leave group behavior} is used.
+ * <p>
+ * This method waits up to the timeout for the consumer to complete pending commits and maybe leaves the group,
Review Comment:
she's right, no s. English is confusing
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2065201574
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated(since = "4.1")
@Override
- @SuppressWarnings("deprecation")
public void close(Duration timeout) {
delegate.close(timeout);
}
/**
- * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.
- * The thread which is blocking in an operation will throw {@link org.apache.kafka.common.errors.WakeupException}.
- * If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
+ * Close the consumer cleanly. {@link CloseOptions} allows to specify a timeout and a
+ * {@link CloseOptions.GroupMembershipOperation leave group behavior}.
+ * If no timeout is specified, the default timeout of 30 seconds is used.
+ * If no leave group behavior is specified, the {@link CloseOptions.GroupMembershipOperation#DEFAULT default
+ * leave group behavior} is used.
+ * <p>
+ * This method waits up to the timeout for the consumer to complete pending commits and maybe leaves the group,
Review Comment:
Thanks for explaining. "Surprisingly", I could follow and it make sense to me 😂 -- Mr.John (english major) @vvcephei could not have explained it any better.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2059376694
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1776,10 +1777,13 @@ public void close() {
}
/**
- * Tries to close the consumer cleanly within the specified timeout. This method waits up to
- * {@code timeout} for the consumer to complete pending commits and leave the group.
+ * This method has been deprecated since Kafka 4.0 and should use {@link KafkaConsumer#close(CloseOptions)} instead.
Review Comment:
```suggestion
* This method has been deprecated since Kafka 4.1 and should use {@link KafkaConsumer#close(CloseOptions)} instead.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2061032584
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated(since = "4.1")
@Override
- @SuppressWarnings("deprecation")
public void close(Duration timeout) {
delegate.close(timeout);
}
/**
- * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.
- * The thread which is blocking in an operation will throw {@link org.apache.kafka.common.errors.WakeupException}.
- * If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
+ * Close the consumer cleanly. {@link CloseOptions} allows to specify a timeout and a
+ * {@link CloseOptions.GroupMembershipOperation leave group behavior}.
+ * If no timeout is specified, the default timeout of 30 seconds is used.
+ * If no leave group behavior is specified, the {@link CloseOptions.GroupMembershipOperation#DEFAULT default
+ * leave group behavior} is used.
+ * <p>
+ * This method waits up to the timeout for the consumer to complete pending commits and maybe leaves the group,
Review Comment:
Cannot follow.
`This method wait[s] ... and maybe leave[s] the group.` -- Why would we remove the `s`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2061081504
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated(since = "4.1")
@Override
- @SuppressWarnings("deprecation")
public void close(Duration timeout) {
delegate.close(timeout);
}
/**
- * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.
- * The thread which is blocking in an operation will throw {@link org.apache.kafka.common.errors.WakeupException}.
- * If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
+ * Close the consumer cleanly. {@link CloseOptions} allows to specify a timeout and a
+ * {@link CloseOptions.GroupMembershipOperation leave group behavior}.
+ * If no timeout is specified, the default timeout of 30 seconds is used.
+ * If no leave group behavior is specified, the {@link CloseOptions.GroupMembershipOperation#DEFAULT default
+ * leave group behavior} is used.
+ * <p>
+ * This method waits up to the timeout for the consumer to complete pending commits and maybe leaves the group,
Review Comment:
To make matters even more confusing, I noticed we actually do _not_ pass the timeout into the consumer or producer's `#close` methods...
So technically-technically the original sentence is also correct code-wise, but imo that's kind of a bug
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2059376028
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated
@Override
- @SuppressWarnings("deprecation")
public void close(Duration timeout) {
delegate.close(timeout);
}
/**
- * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.
- * The thread which is blocking in an operation will throw {@link org.apache.kafka.common.errors.WakeupException}.
- * If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
+ * Close the consumer cleanly. {@link CloseOptions} allows to specify a timeout and a
+ * {@link CloseOptions.GroupMembershipOperation membership operation}.
+ * If no timeout is specified, the default timeout of 30 seconds is uses.
+ * If no membership operation is specified, the {@link CloseOptions.GroupMembershipOperation#DEFAULT default
+ * membership operation} is used.
+ * <p>
+ * This method waits up to the timeout for the consumer to complete pending commits and may leave the group,
+ * depending on the specified membership operation.
+ * If auto-commit is enabled, this will commit the current offsets if possible within the
+ * timeout. If the consumer is unable to complete offset commits and gracefully leave the group (if applicable)
+ * before the timeout expires, the consumer is force closed. Note that {@link #wakeup()} cannot be
+ * used to interrupt close.
+ * <p>
+ * The actual maximum wait time is bounded by the {@link ConsumerConfig#REQUEST_TIMEOUT_MS_CONFIG} setting, which
+ * only applies to operations performed with the broker (coordinator-related requests and
+ * fetch sessions). Even if a larger timeout is specified, the consumer will not wait longer than
+ * {@link ConsumerConfig#REQUEST_TIMEOUT_MS_CONFIG} for these requests to complete during the close operation.
+ * Note that the execution time of callbacks (such as {@link OffsetCommitCallback} and
+ * {@link ConsumerRebalanceListener}) does not consume time from the close timeout.
+ *
+ * @param option see {@link CloseOptions}; cannot be {@code null}
Review Comment:
yeah lets add a check and throw an IAException
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2059538300
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated
Review Comment:
My bad 😅
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2060702298
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated
@Override
- @SuppressWarnings("deprecation")
public void close(Duration timeout) {
delegate.close(timeout);
}
/**
- * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.
- * The thread which is blocking in an operation will throw {@link org.apache.kafka.common.errors.WakeupException}.
- * If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
+ * Close the consumer cleanly. {@link CloseOptions} allows to specify a timeout and a
+ * {@link CloseOptions.GroupMembershipOperation membership operation}.
+ * If no timeout is specified, the default timeout of 30 seconds is uses.
+ * If no membership operation is specified, the {@link CloseOptions.GroupMembershipOperation#DEFAULT default
+ * membership operation} is used.
+ * <p>
+ * This method waits up to the timeout for the consumer to complete pending commits and may leave the group,
+ * depending on the specified membership operation.
+ * If auto-commit is enabled, this will commit the current offsets if possible within the
+ * timeout. If the consumer is unable to complete offset commits and gracefully leave the group (if applicable)
+ * before the timeout expires, the consumer is force closed. Note that {@link #wakeup()} cannot be
+ * used to interrupt close.
+ * <p>
+ * The actual maximum wait time is bounded by the {@link ConsumerConfig#REQUEST_TIMEOUT_MS_CONFIG} setting, which
+ * only applies to operations performed with the broker (coordinator-related requests and
+ * fetch sessions). Even if a larger timeout is specified, the consumer will not wait longer than
+ * {@link ConsumerConfig#REQUEST_TIMEOUT_MS_CONFIG} for these requests to complete during the close operation.
+ * Note that the execution time of callbacks (such as {@link OffsetCommitCallback} and
+ * {@link ConsumerRebalanceListener}) does not consume time from the close timeout.
+ *
+ * @param option see {@link CloseOptions}; cannot be {@code null}
Review Comment:
yeah, I personally prefer IAE too, but we do have this same NPE behaviour on args in many other consumer APIs too (ex. commit, position, committed,...I almost all)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2057700793
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated
@Override
- @SuppressWarnings("deprecation")
public void close(Duration timeout) {
delegate.close(timeout);
}
/**
- * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.
- * The thread which is blocking in an operation will throw {@link org.apache.kafka.common.errors.WakeupException}.
- * If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
+ * Close the consumer cleanly. {@link CloseOptions} allows to specify a timeout and a
+ * {@link CloseOptions.GroupMembershipOperation membership operation}.
+ * If no timeout is specified, the default timeout of 30 seconds is uses.
Review Comment:
```suggestion
* If no timeout is specified, the default timeout of 30 seconds is used.
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated
@Override
- @SuppressWarnings("deprecation")
public void close(Duration timeout) {
delegate.close(timeout);
}
/**
- * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.
- * The thread which is blocking in an operation will throw {@link org.apache.kafka.common.errors.WakeupException}.
- * If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
+ * Close the consumer cleanly. {@link CloseOptions} allows to specify a timeout and a
+ * {@link CloseOptions.GroupMembershipOperation membership operation}.
+ * If no timeout is specified, the default timeout of 30 seconds is uses.
+ * If no membership operation is specified, the {@link CloseOptions.GroupMembershipOperation#DEFAULT default
+ * membership operation} is used.
+ * <p>
+ * This method waits up to the timeout for the consumer to complete pending commits and may leave the group,
+ * depending on the specified membership operation.
+ * If auto-commit is enabled, this will commit the current offsets if possible within the
+ * timeout. If the consumer is unable to complete offset commits and gracefully leave the group (if applicable)
Review Comment:
```suggestion
* timeout. If the consumer is unable to complete offset commits and gracefully leaves the group (if applicable)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2061027260
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated
@Override
- @SuppressWarnings("deprecation")
public void close(Duration timeout) {
delegate.close(timeout);
}
/**
- * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.
- * The thread which is blocking in an operation will throw {@link org.apache.kafka.common.errors.WakeupException}.
- * If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
+ * Close the consumer cleanly. {@link CloseOptions} allows to specify a timeout and a
+ * {@link CloseOptions.GroupMembershipOperation membership operation}.
+ * If no timeout is specified, the default timeout of 30 seconds is uses.
+ * If no membership operation is specified, the {@link CloseOptions.GroupMembershipOperation#DEFAULT default
+ * membership operation} is used.
+ * <p>
+ * This method waits up to the timeout for the consumer to complete pending commits and may leave the group,
+ * depending on the specified membership operation.
+ * If auto-commit is enabled, this will commit the current offsets if possible within the
+ * timeout. If the consumer is unable to complete offset commits and gracefully leave the group (if applicable)
+ * before the timeout expires, the consumer is force closed. Note that {@link #wakeup()} cannot be
+ * used to interrupt close.
+ * <p>
+ * The actual maximum wait time is bounded by the {@link ConsumerConfig#REQUEST_TIMEOUT_MS_CONFIG} setting, which
+ * only applies to operations performed with the broker (coordinator-related requests and
+ * fetch sessions). Even if a larger timeout is specified, the consumer will not wait longer than
+ * {@link ConsumerConfig#REQUEST_TIMEOUT_MS_CONFIG} for these requests to complete during the close operation.
+ * Note that the execution time of callbacks (such as {@link OffsetCommitCallback} and
+ * {@link ConsumerRebalanceListener}) does not consume time from the close timeout.
+ *
+ * @param option see {@link CloseOptions}; cannot be {@code null}
Review Comment:
Ok, so the verdict is we keep NPE. Thanks.
(Btw: also aligns to what KS does.)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax merged PR #19546:
URL: https://github.com/apache/kafka/pull/19546
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2057790431
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated
@Override
- @SuppressWarnings("deprecation")
public void close(Duration timeout) {
delegate.close(timeout);
}
/**
- * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.
- * The thread which is blocking in an operation will throw {@link org.apache.kafka.common.errors.WakeupException}.
- * If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
+ * Close the consumer cleanly. {@link CloseOptions} allows to specify a timeout and a
+ * {@link CloseOptions.GroupMembershipOperation membership operation}.
+ * If no timeout is specified, the default timeout of 30 seconds is uses.
+ * If no membership operation is specified, the {@link CloseOptions.GroupMembershipOperation#DEFAULT default
+ * membership operation} is used.
+ * <p>
+ * This method waits up to the timeout for the consumer to complete pending commits and may leave the group,
+ * depending on the specified membership operation.
+ * If auto-commit is enabled, this will commit the current offsets if possible within the
+ * timeout. If the consumer is unable to complete offset commits and gracefully leave the group (if applicable)
+ * before the timeout expires, the consumer is force closed. Note that {@link #wakeup()} cannot be
+ * used to interrupt close.
+ * <p>
+ * The actual maximum wait time is bounded by the {@link ConsumerConfig#REQUEST_TIMEOUT_MS_CONFIG} setting, which
+ * only applies to operations performed with the broker (coordinator-related requests and
+ * fetch sessions). Even if a larger timeout is specified, the consumer will not wait longer than
+ * {@link ConsumerConfig#REQUEST_TIMEOUT_MS_CONFIG} for these requests to complete during the close operation.
+ * Note that the execution time of callbacks (such as {@link OffsetCommitCallback} and
+ * {@link ConsumerRebalanceListener}) does not consume time from the close timeout.
+ *
+ * @param option see {@link CloseOptions}; cannot be {@code null}
Review Comment:
We don't check whether `CloseOptions` is null. If users use `close((CloseOptions) null)`, they get `NullPointerException`. Do we want to return `IllegalArgumentException` when input is null here? Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2058455119
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated
@Override
- @SuppressWarnings("deprecation")
public void close(Duration timeout) {
delegate.close(timeout);
}
/**
- * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.
- * The thread which is blocking in an operation will throw {@link org.apache.kafka.common.errors.WakeupException}.
- * If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
+ * Close the consumer cleanly. {@link CloseOptions} allows to specify a timeout and a
+ * {@link CloseOptions.GroupMembershipOperation membership operation}.
+ * If no timeout is specified, the default timeout of 30 seconds is uses.
+ * If no membership operation is specified, the {@link CloseOptions.GroupMembershipOperation#DEFAULT default
Review Comment:
as above, would ~"leave group behaviour" be clearer than the generic "membership operation"?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1776,10 +1777,13 @@ public void close() {
}
/**
- * Tries to close the consumer cleanly within the specified timeout. This method waits up to
- * {@code timeout} for the consumer to complete pending commits and leave the group.
+ * This method has been deprecated since Kafka 4.0 and should use {@link KafkaConsumer#close(CloseOptions)} instead.
+ * <p>
+ * Close the consumer with {@link CloseOptions.GroupMembershipOperation#DEFAULT default membership operation}
Review Comment:
"membership operation" seems to generic to me really, this is only about "leave group behaviour" or similar. Would that be clearer as java doc?
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated
Review Comment:
since 4.1 really (helpful adding the since btw)
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated
@Override
- @SuppressWarnings("deprecation")
public void close(Duration timeout) {
delegate.close(timeout);
}
/**
- * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.
- * The thread which is blocking in an operation will throw {@link org.apache.kafka.common.errors.WakeupException}.
- * If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
+ * Close the consumer cleanly. {@link CloseOptions} allows to specify a timeout and a
+ * {@link CloseOptions.GroupMembershipOperation membership operation}.
+ * If no timeout is specified, the default timeout of 30 seconds is uses.
+ * If no membership operation is specified, the {@link CloseOptions.GroupMembershipOperation#DEFAULT default
+ * membership operation} is used.
+ * <p>
+ * This method waits up to the timeout for the consumer to complete pending commits and may leave the group,
Review Comment:
```suggestion
* This method waits up to the timeout for the consumer to complete pending commits and maybe leave the group,
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1776,10 +1777,13 @@ public void close() {
}
/**
- * Tries to close the consumer cleanly within the specified timeout. This method waits up to
- * {@code timeout} for the consumer to complete pending commits and leave the group.
+ * This method has been deprecated since Kafka 4.0 and should use {@link KafkaConsumer#close(CloseOptions)} instead.
+ * <p>
+ * Close the consumer with {@link CloseOptions.GroupMembershipOperation#DEFAULT default membership operation}
+ * cleanly within the specified timeout. This method waits up to
+ * {@code timeout} for the consumer to complete pending commits and may leave the group (if applicable).
Review Comment:
about the "and may leave the group (if applicable).", could we be explicit in this case, just to make things clear? (it will leave the group if it's a dynamic member). That's the only case here that the close has the DEFAULT membership behaviour I expect.
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated
@Override
- @SuppressWarnings("deprecation")
public void close(Duration timeout) {
delegate.close(timeout);
}
/**
- * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.
- * The thread which is blocking in an operation will throw {@link org.apache.kafka.common.errors.WakeupException}.
- * If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
+ * Close the consumer cleanly. {@link CloseOptions} allows to specify a timeout and a
+ * {@link CloseOptions.GroupMembershipOperation membership operation}.
+ * If no timeout is specified, the default timeout of 30 seconds is uses.
+ * If no membership operation is specified, the {@link CloseOptions.GroupMembershipOperation#DEFAULT default
+ * membership operation} is used.
+ * <p>
+ * This method waits up to the timeout for the consumer to complete pending commits and may leave the group,
+ * depending on the specified membership operation.
+ * If auto-commit is enabled, this will commit the current offsets if possible within the
+ * timeout. If the consumer is unable to complete offset commits and gracefully leave the group (if applicable)
Review Comment:
yeah, a bit confusing, but in this case "leave" is correct, referring to "unable to ...leave"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2059527520
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated
Review Comment:
Well, 4.1 :) -- c&p error form the previous PR...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2059377983
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated
@Override
- @SuppressWarnings("deprecation")
public void close(Duration timeout) {
delegate.close(timeout);
}
/**
- * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.
- * The thread which is blocking in an operation will throw {@link org.apache.kafka.common.errors.WakeupException}.
- * If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
+ * Close the consumer cleanly. {@link CloseOptions} allows to specify a timeout and a
+ * {@link CloseOptions.GroupMembershipOperation membership operation}.
+ * If no timeout is specified, the default timeout of 30 seconds is uses.
+ * If no membership operation is specified, the {@link CloseOptions.GroupMembershipOperation#DEFAULT default
Review Comment:
agreed, would also be good to specify what the default leave behavior actually is
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #19546:
URL: https://github.com/apache/kafka/pull/19546#issuecomment-2830983993
Updates this PR. Call for review.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2060947545
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated
@Override
- @SuppressWarnings("deprecation")
public void close(Duration timeout) {
delegate.close(timeout);
}
/**
- * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.
- * The thread which is blocking in an operation will throw {@link org.apache.kafka.common.errors.WakeupException}.
- * If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
+ * Close the consumer cleanly. {@link CloseOptions} allows to specify a timeout and a
+ * {@link CloseOptions.GroupMembershipOperation membership operation}.
+ * If no timeout is specified, the default timeout of 30 seconds is uses.
+ * If no membership operation is specified, the {@link CloseOptions.GroupMembershipOperation#DEFAULT default
+ * membership operation} is used.
+ * <p>
+ * This method waits up to the timeout for the consumer to complete pending commits and may leave the group,
+ * depending on the specified membership operation.
+ * If auto-commit is enabled, this will commit the current offsets if possible within the
+ * timeout. If the consumer is unable to complete offset commits and gracefully leave the group (if applicable)
+ * before the timeout expires, the consumer is force closed. Note that {@link #wakeup()} cannot be
+ * used to interrupt close.
+ * <p>
+ * The actual maximum wait time is bounded by the {@link ConsumerConfig#REQUEST_TIMEOUT_MS_CONFIG} setting, which
+ * only applies to operations performed with the broker (coordinator-related requests and
+ * fetch sessions). Even if a larger timeout is specified, the consumer will not wait longer than
+ * {@link ConsumerConfig#REQUEST_TIMEOUT_MS_CONFIG} for these requests to complete during the close operation.
+ * Note that the execution time of callbacks (such as {@link OffsetCommitCallback} and
+ * {@link ConsumerRebalanceListener}) does not consume time from the close timeout.
+ *
+ * @param option see {@link CloseOptions}; cannot be {@code null}
Review Comment:
fair enough
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2059377623
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1776,10 +1777,13 @@ public void close() {
}
/**
- * Tries to close the consumer cleanly within the specified timeout. This method waits up to
- * {@code timeout} for the consumer to complete pending commits and leave the group.
+ * This method has been deprecated since Kafka 4.0 and should use {@link KafkaConsumer#close(CloseOptions)} instead.
+ * <p>
+ * Close the consumer with {@link CloseOptions.GroupMembershipOperation#DEFAULT default membership operation}
+ * cleanly within the specified timeout. This method waits up to
+ * {@code timeout} for the consumer to complete pending commits and may leave the group (if applicable).
* If auto-commit is enabled, this will commit the current offsets if possible within the
- * timeout. If the consumer is unable to complete offset commits and gracefully leave the group
+ * timeout. If the consumer is unable to complete offset commits and gracefully leave the group (if applicable)
Review Comment:
(if applicable) is ok to use here as long as we specified when it's applicable earlier in the javadocs
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2059527767
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated
@Override
- @SuppressWarnings("deprecation")
public void close(Duration timeout) {
delegate.close(timeout);
}
/**
- * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.
- * The thread which is blocking in an operation will throw {@link org.apache.kafka.common.errors.WakeupException}.
- * If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
+ * Close the consumer cleanly. {@link CloseOptions} allows to specify a timeout and a
+ * {@link CloseOptions.GroupMembershipOperation membership operation}.
+ * If no timeout is specified, the default timeout of 30 seconds is uses.
+ * If no membership operation is specified, the {@link CloseOptions.GroupMembershipOperation#DEFAULT default
+ * membership operation} is used.
+ * <p>
+ * This method waits up to the timeout for the consumer to complete pending commits and may leave the group,
+ * depending on the specified membership operation.
+ * If auto-commit is enabled, this will commit the current offsets if possible within the
+ * timeout. If the consumer is unable to complete offset commits and gracefully leave the group (if applicable)
+ * before the timeout expires, the consumer is force closed. Note that {@link #wakeup()} cannot be
+ * used to interrupt close.
+ * <p>
+ * The actual maximum wait time is bounded by the {@link ConsumerConfig#REQUEST_TIMEOUT_MS_CONFIG} setting, which
+ * only applies to operations performed with the broker (coordinator-related requests and
+ * fetch sessions). Even if a larger timeout is specified, the consumer will not wait longer than
+ * {@link ConsumerConfig#REQUEST_TIMEOUT_MS_CONFIG} for these requests to complete during the close operation.
+ * Note that the execution time of callbacks (such as {@link OffsetCommitCallback} and
+ * {@link ConsumerRebalanceListener}) does not consume time from the close timeout.
+ *
+ * @param option see {@link CloseOptions}; cannot be {@code null}
Review Comment:
I think it's ok if users get NPE -- we say "cannot be `null`" so NPE should be expected?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2060705096
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1776,10 +1777,13 @@ public void close() {
}
/**
- * Tries to close the consumer cleanly within the specified timeout. This method waits up to
- * {@code timeout} for the consumer to complete pending commits and leave the group.
+ * This method has been deprecated since Kafka 4.1 and should use {@link KafkaConsumer#close(CloseOptions)} instead.
+ * <p>
+ * Close the consumer with {@link CloseOptions.GroupMembershipOperation#DEFAULT default leave group behavior}
+ * cleanly within the specified timeout. This method waits up to
+ * {@code timeout} for the consumer to complete pending commits and maybe leaves the group (if the group is dynamic).
Review Comment:
```suggestion
* {@code timeout} for the consumer to complete pending commits and maybe leave the group (if the member is dynamic).
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated(since = "4.1")
@Override
- @SuppressWarnings("deprecation")
public void close(Duration timeout) {
delegate.close(timeout);
}
/**
- * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.
- * The thread which is blocking in an operation will throw {@link org.apache.kafka.common.errors.WakeupException}.
- * If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
+ * Close the consumer cleanly. {@link CloseOptions} allows to specify a timeout and a
+ * {@link CloseOptions.GroupMembershipOperation leave group behavior}.
+ * If no timeout is specified, the default timeout of 30 seconds is used.
+ * If no leave group behavior is specified, the {@link CloseOptions.GroupMembershipOperation#DEFAULT default
+ * leave group behavior} is used.
+ * <p>
+ * This method waits up to the timeout for the consumer to complete pending commits and maybe leaves the group,
Review Comment:
```suggestion
* This method waits up to the timeout for the consumer to complete pending commits and maybe leave the group,
```
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated(since = "4.1")
@Override
- @SuppressWarnings("deprecation")
public void close(Duration timeout) {
delegate.close(timeout);
}
/**
- * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.
- * The thread which is blocking in an operation will throw {@link org.apache.kafka.common.errors.WakeupException}.
- * If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
+ * Close the consumer cleanly. {@link CloseOptions} allows to specify a timeout and a
+ * {@link CloseOptions.GroupMembershipOperation leave group behavior}.
+ * If no timeout is specified, the default timeout of 30 seconds is used.
+ * If no leave group behavior is specified, the {@link CloseOptions.GroupMembershipOperation#DEFAULT default
+ * leave group behavior} is used.
+ * <p>
+ * This method waits up to the timeout for the consumer to complete pending commits and maybe leaves the group,
+ * depending on the specified leave group behavior.
+ * If auto-commit is enabled, this will commit the current offsets if possible within the
+ * timeout. If the consumer is unable to complete offset commits and gracefully leaves the group (if applicable)
Review Comment:
```suggestion
* timeout. If the consumer is unable to complete offset commits and gracefully leave the group (if applicable)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2061081297
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated(since = "4.1")
@Override
- @SuppressWarnings("deprecation")
public void close(Duration timeout) {
delegate.close(timeout);
}
/**
- * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.
- * The thread which is blocking in an operation will throw {@link org.apache.kafka.common.errors.WakeupException}.
- * If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
+ * Close the consumer cleanly. {@link CloseOptions} allows to specify a timeout and a
+ * {@link CloseOptions.GroupMembershipOperation leave group behavior}.
+ * If no timeout is specified, the default timeout of 30 seconds is used.
+ * If no leave group behavior is specified, the {@link CloseOptions.GroupMembershipOperation#DEFAULT default
+ * leave group behavior} is used.
+ * <p>
+ * This method waits up to the timeout for the consumer to complete pending commits and maybe leaves the group,
Review Comment:
technically the original sentence is also grammatically correct, however it doesn't mean exactly the same thing.
With the s it's is effectively saying "This method (waits up to the timeout for the consumer to complete pending commits) and maybe leaves the group" which implies leaving the group isn't part of the timeout.
Without the s it's grouped as "this method waits up to the timeout for the consumer to (complete pending commits and maybe leave the group)" which implies leaving the group is part of the timeout
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "ableegoldman (via GitHub)" <gi...@apache.org>.
ableegoldman commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2061079573
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated(since = "4.1")
@Override
- @SuppressWarnings("deprecation")
public void close(Duration timeout) {
delegate.close(timeout);
}
/**
- * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.
- * The thread which is blocking in an operation will throw {@link org.apache.kafka.common.errors.WakeupException}.
- * If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
+ * Close the consumer cleanly. {@link CloseOptions} allows to specify a timeout and a
+ * {@link CloseOptions.GroupMembershipOperation leave group behavior}.
+ * If no timeout is specified, the default timeout of 30 seconds is used.
+ * If no leave group behavior is specified, the {@link CloseOptions.GroupMembershipOperation#DEFAULT default
+ * leave group behavior} is used.
+ * <p>
+ * This method waits up to the timeout for the consumer to complete pending commits and maybe leaves the group,
Review Comment:
i don't know the proper grammar terminology here but essentially it's because the "maybe leave the group" is part of the X in the sentence "this method waits up to the timeout for the consumer to X"
here the "maybe leave the group" comes after an "and" which can be amiguous but in this context I think it makes more sense for it to be part of the X, ie "this method waits up to the timeout for the consumer to (complete pending commits and maybe leave the group)"
so its should be conjugated the same way as the thing before the and, ie the "complete pending commits"
since it's "complete" and not "completes", it should also be "leave" and not "leaves"
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "lianetm (via GitHub)" <gi...@apache.org>.
lianetm commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2060702298
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated
@Override
- @SuppressWarnings("deprecation")
public void close(Duration timeout) {
delegate.close(timeout);
}
/**
- * Wakeup the consumer. This method is thread-safe and is useful in particular to abort a long poll.
- * The thread which is blocking in an operation will throw {@link org.apache.kafka.common.errors.WakeupException}.
- * If no thread is blocking in a method which can throw {@link org.apache.kafka.common.errors.WakeupException}, the next call to such a method will raise it instead.
+ * Close the consumer cleanly. {@link CloseOptions} allows to specify a timeout and a
+ * {@link CloseOptions.GroupMembershipOperation membership operation}.
+ * If no timeout is specified, the default timeout of 30 seconds is uses.
+ * If no membership operation is specified, the {@link CloseOptions.GroupMembershipOperation#DEFAULT default
+ * membership operation} is used.
+ * <p>
+ * This method waits up to the timeout for the consumer to complete pending commits and may leave the group,
+ * depending on the specified membership operation.
+ * If auto-commit is enabled, this will commit the current offsets if possible within the
+ * timeout. If the consumer is unable to complete offset commits and gracefully leave the group (if applicable)
+ * before the timeout expires, the consumer is force closed. Note that {@link #wakeup()} cannot be
+ * used to interrupt close.
+ * <p>
+ * The actual maximum wait time is bounded by the {@link ConsumerConfig#REQUEST_TIMEOUT_MS_CONFIG} setting, which
+ * only applies to operations performed with the broker (coordinator-related requests and
+ * fetch sessions). Even if a larger timeout is specified, the consumer will not wait longer than
+ * {@link ConsumerConfig#REQUEST_TIMEOUT_MS_CONFIG} for these requests to complete during the close operation.
+ * Note that the execution time of callbacks (such as {@link OffsetCommitCallback} and
+ * {@link ConsumerRebalanceListener}) does not consume time from the close timeout.
+ *
+ * @param option see {@link CloseOptions}; cannot be {@code null}
Review Comment:
yeah, I personally prefer IAE too, but we do have this same NPE behaviour on args in many other consumer APIs too (ex. commit, position, committed,...I think almost all)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: improve JavaDocs for consumer CloseOptions [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19546:
URL: https://github.com/apache/kafka/pull/19546#discussion_r2057755179
##########
clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java:
##########
@@ -1797,31 +1801,48 @@ public void close() {
* @throws InterruptException If the thread is interrupted before or while this function is called
* @throws org.apache.kafka.common.KafkaException for any other error during close
*/
+ @Deprecated
Review Comment:
```suggestion
@Deprecated(since = "4.0")
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org