frankvicky commented on code in PR #22405:
URL: https://github.com/apache/kafka/pull/22405#discussion_r3631274207
##########
clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerHeartbeatRequestManager.java:
##########
@@ -100,12 +101,14 @@ public boolean handleSpecificFailure(Throwable exception)
{
String errorMessage = exception.getMessage();
if (exception instanceof UnsupportedVersionException) {
String message = CONSUMER_PROTOCOL_NOT_SUPPORTED_MSG;
- if (errorMessage.equals(REGEX_RESOLUTION_NOT_SUPPORTED_MSG)) {
+ if (exception instanceof UnsupportedProtocolFieldException) {
message = REGEX_RESOLUTION_NOT_SUPPORTED_MSG;
logger.error("{} regex resolution not supported: {}",
heartbeatRequestName(), message);
Review Comment:
Could we use `exception.getMessage()` here (and in the log line above)? That
way we don't need to touch this branch when a new field validation is added to
`ConsumerGroupHeartbeatRequest.Builder` — the specific message from the builder
will flow through automatically. Also, `errorMessage` on line 101 is currently
unused in this branch, so this would put it back to work.
##########
clients/src/main/java/org/apache/kafka/common/requests/ElectLeadersRequest.java:
##########
@@ -63,7 +63,7 @@ public String toString() {
private ElectLeadersRequestData toRequestData(short version) {
if (electionType != ElectionType.PREFERRED && version == 0) {
- throw new UnsupportedVersionException("API Version 0 only
supports PREFERRED election type");
+ throw new UnsupportedProtocolFieldException("ElectionType",
apiKey().name(), version, 1);
Review Comment:
Could it be better?
```suggestion
throw new
UnsupportedProtocolFieldException(electionType.name(), apiKey().name(),
version, 1);
```
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java:
##########
@@ -575,17 +575,14 @@ private Set<String> createTopics(final Set<NewTopic>
topicsToCreate,
log.error("Unexpected error during topic creation for
{}.\n" +
"Error message was: {}", topicName,
cause.toString());
- if (cause instanceof UnsupportedVersionException) {
- final String errorMessage = cause.getMessage();
- if (errorMessage != null &&
- errorMessage.startsWith("Creating topics with
default partitions/replication factor are only supported in CreateTopicRequest
version 4+")) {
-
- throw new StreamsException(String.format(
- "Could not create topic %s, because
brokers don't support configuration replication.factor=-1."
- + " You can change the
replication.factor config or upgrade your brokers to version 2.4 or newer to
avoid this error.",
- topicName)
- );
- }
+ if (cause instanceof UnsupportedProtocolFieldException) {
+ // An older broker rejected a field we rely on (e.g.
the default
+ // replication.factor=-1, which requires CreateTopics
request version 4+).
+ throw new StreamsException(String.format(
+ "Could not create topic %s, because brokers
don't support configuration replication.factor=-1."
+ + " You can change the
replication.factor config or upgrade your brokers to version 2.4 or newer to
avoid this error.",
+ topicName)
+ );
Review Comment:
This branch now matches any `UnsupportedProtocolFieldException`, but the
`StreamsException` message still hard-codes `replication.factor=-1` as the root
cause. Today `CreateTopicsRequest.Builder` throws this exception in two places
(`validateOnly` at v0, and `topicsWithDefaults` at v<4), and Streams only hits
the latter — so it works for now.
But if a new field validation is ever added to the builder, we'll
misattribute the failure here.
Could we pass `cause.getMessage()` into the `StreamsException` instead of
the fixed message?
That keeps the type-based check (which matches the spirit of this PR)
without hard-coding an assumption about which field failed.
##########
clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java:
##########
@@ -699,8 +700,7 @@ public void
testCreateTopicRequestV3FailsIfNoPartitionsOrReplicas() {
new Builder(data).build((short) 3);
});
- assertTrue(exception.getMessage().contains("supported in
CreateTopicRequest version 4+"));
- assertTrue(exception.getMessage().contains("[foo, bar]"));
+ assertTrue(exception.getMessage().contains("does not support [foo,bar]
in CREATE_TOPICS API version 3"));
Review Comment:
Should we add a unit test for `UnsupportedProtocolFieldException`?
##########
clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java:
##########
@@ -52,10 +51,8 @@ public CreateTopicsRequest build(short version) {
.collect(Collectors.toList());
if (!topicsWithDefaults.isEmpty() && version < 4) {
- throw new UnsupportedVersionException("Creating topics with
default "
- + "partitions/replication factor are only supported in
CreateTopicRequest "
- + "version 4+. The following topics need values for
partitions and replicas: "
- + topicsWithDefaults);
+ throw new UnsupportedProtocolFieldException(String.join(",",
topicsWithDefaults),
+ apiKey().name(), version, 4);
Review Comment:
Passing the topic names as `fieldOrValue` makes the message read as "the
cluster does not support [foo,bar] in CREATE_TOPICS API version 3", which
sounds like the topic names themselves are unsupported. The actual unsupported
thing is using default partitions/replication for those topics. Could we pass
something like `"default partitions/replication for topics [" +
String.join(",", topicsWithDefaults) + "]"` instead, so the message accurately
describes what the older broker rejected?
##########
clients/src/main/java/org/apache/kafka/common/requests/ListGroupsRequest.java:
##########
@@ -60,9 +59,7 @@ public ListGroupsRequest build(short version) {
boolean containedClassic =
typesCopy.remove(GroupType.CLASSIC.toString());
boolean containedConsumer =
typesCopy.remove(GroupType.CONSUMER.toString());
if (!typesCopy.isEmpty() || (!containedClassic &&
containedConsumer)) {
- throw new UnsupportedVersionException("The broker only
supports ListGroups " +
- "v" + version + ", but we need v5 or newer to request
groups by type. " +
- "Requested group types: [" + String.join(", ",
data.typesFilter()) + "].");
+ throw new UnsupportedProtocolFieldException("TypesFilter",
apiKey().name(), version, 5);
Review Comment:
The new message drops the actual `data.typesFilter()` contents, which the
old message included ("Requested group types: [...]"). Without it the user
can't tell which of their requested types the broker rejected. Could we pass
the offending types into the field argument — e.g. the `typesFilter` values
that remain after removing the versions-supported ones (`CLASSIC`/`CONSUMER`) —
so the message still shows what actually failed?
##########
clients/src/main/java/org/apache/kafka/common/requests/CreateAclsRequest.java:
##########
@@ -90,8 +92,13 @@ private void validate(CreateAclsRequestData data) {
if (version() == 0) {
final boolean unsupported =
data.creations().stream().anyMatch(creation ->
creation.resourcePatternType() != PatternType.LITERAL.code());
- if (unsupported)
- throw new UnsupportedVersionException("Version 0 only supports
literal resource pattern types");
+ if (unsupported) {
+ String unsupportedType = Arrays.stream(PatternType.values())
+ .filter(type -> type != PatternType.LITERAL)
+ .map(PatternType::name)
+ .collect(Collectors.joining(","));
+ throw new UnsupportedProtocolFieldException(unsupportedType,
apiKey().name(), version(), 1);
+ }
Review Comment:
The `unsupportedType` string is built by joining every non-LITERAL value
from `PatternType.values()`, so the message will claim the cluster doesn't
support that whole list even when the request only contained one non-LITERAL
type. Could we surface the actual offending pattern type from the request (e.g.
`creation.resourcePatternType()`) instead? The old message ("only LITERAL and
ANY are supported" / "only literal resource pattern types") was also more
actionable — it told the user what IS supported rather than enumerating what
isn't. Same suggestion applies to `DeleteAclsRequest` and `DescribeAclsRequest`.
--
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]
