AndrewJSchofield commented on code in PR #19113:
URL: https://github.com/apache/kafka/pull/19113#discussion_r1983000104
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java:
##########
@@ -1062,6 +1065,24 @@ private void ensureExplicitAcknowledgement() {
}
}
+ /**
+ * Initializes the acknowledgement mode based on the configuration.
+ */
+ private static AcknowledgementMode
initializeAcknowledgementMode(ConsumerConfig config) {
+ if (config == null) {
+ return AcknowledgementMode.UNKNOWN;
+ }
+ String acknowledgementModeStr =
config.getString(ConsumerConfig.INTERNAL_SHARE_ACKNOWLEDGEMENT_MODE_CONFIG);
+ if ((acknowledgementModeStr == null) ||
acknowledgementModeStr.isEmpty()) {
+ return AcknowledgementMode.UNKNOWN;
+ } else if (acknowledgementModeStr.equalsIgnoreCase("implicit")) {
+ return AcknowledgementMode.IMPLICIT;
+ } else if (acknowledgementModeStr.equalsIgnoreCase("explicit")) {
+ return AcknowledgementMode.EXPLICIT;
+ }
+ return AcknowledgementMode.UNKNOWN;
Review Comment:
In theory, this should not happen because of the config validation. But,
I'll add it in this PR, and review this whole area when the finalised config is
added. Because the config is allowed for share consumers and not allowed for
regular consumers, it's a bit tricky to make sure that it validates correctly
for both types of consumer.
--
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]
