dajac commented on code in PR #15211:
URL: https://github.com/apache/kafka/pull/15211#discussion_r1469217586
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/classic/ClassicGroup.java:
##########
@@ -956,6 +956,20 @@ public Optional<OffsetExpirationCondition>
offsetExpirationCondition() {
return Optional.empty();
}
+ /**
+ * The state of this group.
+ *
+ * @return The current state as a lowercase String.
+ */
+ public String stateAsLowerCaseString() {
+ return this.state.toLowerCaseString();
+ }
+
+ @Override
+ public boolean isInStates(Set<String> statesFilter, long committedOffset) {
+ return statesFilter.contains(this.stateAsLowerCaseString()) ||
statesFilter.contains(this.stateAsString());
Review Comment:
I am not sure to understand why do we need to check both versions here? If
the `statesFilter` are lowercased, the first one should be enough, no?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -67,14 +68,22 @@ public enum ConsumerGroupState {
private final String name;
+ private final String capitalizedName;
Review Comment:
I would rather do it the other way around in order to be consistent with the
classic group states. Could we capitalize the states in L63-67 and lower case
them here?
##########
core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala:
##########
@@ -841,6 +842,10 @@ private[group] class GroupMetadata(val groupId: String,
initialState: GroupState
.format(groupId, targetState.validPreviousStates.mkString(","),
targetState, state))
}
+ def isInStates(states: collection.Set[String]): Boolean = {
+ states.contains(state.toString) || states.contains(state.toLowerCaseString)
Review Comment:
Same question.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -739,6 +748,19 @@ public Optional<OffsetExpirationCondition>
offsetExpirationCondition() {
return Optional.of(new OffsetExpirationConditionImpl(offsetAndMetadata
-> offsetAndMetadata.commitTimestampMs));
}
+ /**
+ * @return The current state as a capitalized String with given
committedOffset.
+ */
+ public String stateAsCapitalizedString(long committedOffset) {
+ return state.get(committedOffset).toCapitalizedString();
+ }
Review Comment:
Same suggestion about inlining it.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/classic/ClassicGroup.java:
##########
@@ -956,6 +956,20 @@ public Optional<OffsetExpirationCondition>
offsetExpirationCondition() {
return Optional.empty();
}
+ /**
+ * The state of this group.
+ *
+ * @return The current state as a lowercase String.
+ */
+ public String stateAsLowerCaseString() {
+ return this.state.toLowerCaseString();
+ }
Review Comment:
We could probably remove this one and inline it.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -739,6 +748,19 @@ public Optional<OffsetExpirationCondition>
offsetExpirationCondition() {
return Optional.of(new OffsetExpirationConditionImpl(offsetAndMetadata
-> offsetAndMetadata.commitTimestampMs));
}
+ /**
+ * @return The current state as a capitalized String with given
committedOffset.
+ */
+ public String stateAsCapitalizedString(long committedOffset) {
+ return state.get(committedOffset).toCapitalizedString();
+ }
+
+ @Override
+ public boolean isInStates(Set<String> statesFilter, long committedOffset) {
+ return
statesFilter.contains(this.stateAsCapitalizedString(committedOffset)) ||
statesFilter.contains(
+ this.stateAsString(committedOffset));
Review Comment:
Same question about checking the two versions here.
--
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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
