JimmyWang6 commented on code in PR #20246:
URL: https://github.com/apache/kafka/pull/20246#discussion_r2488405362
##########
clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java:
##########
@@ -380,14 +380,25 @@ public class ConsumerConfig extends AbstractConfig {
" If set to <code>explicit</code>, the acknowledgement mode of the
consumer is explicit and it must use" +
"
<code>org.apache.kafka.clients.consumer.ShareConsumer.acknowledge()</code> to
acknowledge delivery of records.";
+ /**
+ * <code>share.acquire.mode</code>
+ */
+ public static final String SHARE_ACQUIRE_MODE_CONFIG =
"share.acquire.mode";
+ private static final String SHARE_ACQUIRE_MODE_DOC = "Controls the acquire
mode for a share consumer." +
+ " If set to <code>record_limit</code>, the number of records
returned in each poll() will not exceed the value of
<code>max.poll.records</code>." +
+ " If set to <code>batch_optimized</code>, the number of records
returned in each poll() call may exceed <code>max.poll.records</code>" +
+ " to align with batch boundaries for optimization.";
+ public static final String DEFAULT_SHARE_ACQUIRE_MODE = "batch_optimized";
Review Comment:
Done.
##########
clients/src/main/java/org/apache/kafka/common/requests/ShareFetchRequest.java:
##########
@@ -143,6 +145,10 @@ public ShareFetchRequest build(short version) {
if (data.isRenewAck()) {
throw new UnsupportedVersionException("The v1 ShareFetch
does not support AcknowledgeType.RENEW");
}
+ // The v1 does not support ShareAcquireMode record_limit.
+ if (data.shareAcquireMode() ==
ShareAcquireMode.RECORD_LIMIT.id()) {
Review Comment:
Done.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
