AndrewJSchofield commented on code in PR #23317: URL: https://github.com/apache/kafka/pull/23317#discussion_r3979734045
########## generator/src/main/java/org/apache/kafka/message/HeaderVersions.java: ########## @@ -0,0 +1,159 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.kafka.message; + +import java.util.ArrayList; +import java.util.Comparator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** + * The mapping from request/response versions to header versions for an RPC. + * + * The entries form an ascending, non-overlapping, contiguous set of version ranges that starts at + * version 0 and ends with an open-ended range, e.g. {@code { "0-1": "1", "2+": "2" }}. Like + * {@code flexibleVersions}, the map covers every version the schema describes, including versions + * that are no longer valid. + */ +public final class HeaderVersions { + public static final class Entry { + private final Versions range; + private final short headerVersion; + + Entry(Versions range, short headerVersion) { + this.range = range; + this.headerVersion = headerVersion; + } + + public Versions range() { + return range; + } + + public short headerVersion() { + return headerVersion; + } + } + + private final List<Entry> entries; + + private HeaderVersions(List<Entry> entries) { + this.entries = entries; + } + + /** + * Parse the raw {@code headerVersions} map from a message schema. + * + * @param messageName the message name, used in error messages + * @param raw the raw map from the schema, or null if the property is absent + * @param validVersions the valid versions of the message + * @return the parsed header versions, or null if {@code raw} is null + */ + public static HeaderVersions parse(String messageName, Map<String, String> raw, Versions validVersions) { + if (raw == null) { + return null; + } + if (raw.isEmpty()) { + throw new RuntimeException("Message " + messageName + " specifies an empty headerVersions map."); + } + List<Entry> entries = new ArrayList<>(); + for (Map.Entry<String, String> entry : raw.entrySet()) { + entries.add(parseEntry(messageName, entry.getKey(), entry.getValue())); + } + entries.sort(Comparator.comparingInt(entry -> entry.range.lowest())); + validate(messageName, entries, validVersions); + return new HeaderVersions(entries); + } + + private static Entry parseEntry(String messageName, String key, String value) { + if (key == null || key.trim().isEmpty()) { + throw new RuntimeException("Message " + messageName + + " specifies a blank version range in headerVersions."); + } + Versions range; + try { + range = Versions.parse(key, null); + } catch (RuntimeException e) { + // Versions.parse throws NumberFormatException for a non-numeric bound and a plain + // RuntimeException for a negative one; both mean the key is not a valid version range. + range = null; + } + if (range == null || range.empty()) { + throw new RuntimeException("Message " + messageName + + " specifies an invalid version range \"" + key + "\" in headerVersions."); + } + if (value == null || value.trim().isEmpty()) { + throw new RuntimeException("Message " + messageName + + " specifies a blank header version for range \"" + key + "\" in headerVersions."); + } + short headerVersion; + try { + headerVersion = Short.parseShort(value.trim()); + } catch (NumberFormatException e) { + throw new RuntimeException("Message " + messageName + " specifies an invalid header version \"" + + value + "\" for range \"" + key + "\" in headerVersions."); + } + if (headerVersion < 0) { + throw new RuntimeException("Message " + messageName + " specifies a negative header version \"" + + value + "\" for range \"" + key + "\" in headerVersions."); + } + return new Entry(range, headerVersion); + } + + private static void validate(String messageName, List<Entry> entries, Versions validVersions) { + if (entries.get(0).range.lowest() != 0) { + throw new RuntimeException("Message " + messageName + " has headerVersions starting at version " + + entries.get(0).range.lowest() + ", but the first range must start at version 0 so that the map " + + "covers every version the schema describes, including versions that are no longer valid."); + } + for (int i = 1; i < entries.size(); i++) { + int expected = entries.get(i - 1).range.highest() + 1; + if (entries.get(i).range.lowest() != expected) { + throw new RuntimeException("Message " + messageName + " has non-contiguous headerVersions: the " + + "range after " + entries.get(i - 1).range + " must start at version " + expected + + ", but it starts at version " + entries.get(i).range.lowest() + "."); + } + } + Entry last = entries.get(entries.size() - 1); + if (last.range.highest() != Short.MAX_VALUE) { Review Comment: The final entry in the headerVersions map must be an open-ended range, but if I accidentally have an open-ended range in a non-final entry, the error message is weird (`Caused by: java.lang.RuntimeException: Message OffsetDeleteResponse has non-contiguous headerVersions: the range after 0+ must start at version 32768, but it starts at version 1.`). ########## generator/src/main/java/org/apache/kafka/message/HeaderVersions.java: ########## @@ -0,0 +1,159 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.kafka.message; + +import java.util.ArrayList; +import java.util.Comparator; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +/** + * The mapping from request/response versions to header versions for an RPC. + * + * The entries form an ascending, non-overlapping, contiguous set of version ranges that starts at + * version 0 and ends with an open-ended range, e.g. {@code { "0-1": "1", "2+": "2" }}. Like + * {@code flexibleVersions}, the map covers every version the schema describes, including versions + * that are no longer valid. + */ +public final class HeaderVersions { + public static final class Entry { + private final Versions range; + private final short headerVersion; + + Entry(Versions range, short headerVersion) { + this.range = range; + this.headerVersion = headerVersion; + } + + public Versions range() { + return range; + } + + public short headerVersion() { + return headerVersion; + } + } + + private final List<Entry> entries; + + private HeaderVersions(List<Entry> entries) { + this.entries = entries; + } + + /** + * Parse the raw {@code headerVersions} map from a message schema. + * + * @param messageName the message name, used in error messages + * @param raw the raw map from the schema, or null if the property is absent + * @param validVersions the valid versions of the message + * @return the parsed header versions, or null if {@code raw} is null + */ + public static HeaderVersions parse(String messageName, Map<String, String> raw, Versions validVersions) { + if (raw == null) { + return null; + } + if (raw.isEmpty()) { + throw new RuntimeException("Message " + messageName + " specifies an empty headerVersions map."); + } + List<Entry> entries = new ArrayList<>(); + for (Map.Entry<String, String> entry : raw.entrySet()) { + entries.add(parseEntry(messageName, entry.getKey(), entry.getValue())); + } + entries.sort(Comparator.comparingInt(entry -> entry.range.lowest())); + validate(messageName, entries, validVersions); Review Comment: We don't check that the header version are valid. For example, even though request header v3 does not yet exist, I can write: ``` "validVersions": "0-6", "flexibleVersions": "3+", "headerVersions": { "0-2": "1", "3-5": "2", "6+": "3" }, ``` Probably worth having a bit more checking in here to help people writing or changing the RPC definitions and making silly mistakes they might not easily spot. -- 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]
