Skip to content

KAFKA-12278; Ensure exposed api versions are consistent within listener scopes #10066

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
hachikuji wants to merge 7 commits into from
Closed

KAFKA-12278; Ensure exposed api versions are consistent within listener scopes #10066

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
696e798
KAFKA-12278; Define api scope
hachikuji Feb 4, 2021
fd268fb
Override `DefaultApiVersionManager.isApiEnabled` with more efficient …
hachikuji Feb 5, 2021
51d00d0
Fix metadata benchmark
hachikuji Feb 5, 2021
e17cd66
Rename `scope` to `listeners`
hachikuji Feb 18, 2021
b287f8b
Remove `raft` listener type
hachikuji Feb 18, 2021
49d73fd
Fix breakage after merge
hachikuji Feb 18, 2021
882c36d
Store static map of apis by listener type
hachikuji Feb 18, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions checkstyle/import-control.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,6 +50,7 @@

<subpackage name="common">
<allow class="org.apache.kafka.clients.consumer.ConsumerRecord" exact-match="true" />
<allow class="org.apache.kafka.common.message.ApiMessageType" exact-match="true" />
<disallow pkg="org.apache.kafka.clients" />
<allow pkg="org.apache.kafka.common" exact-match="true" />
<allow pkg="org.apache.kafka.common.annotation" />
Expand Down Expand Up @@ -108,6 +109,7 @@
<allow pkg="org.apache.kafka.common.config" />
<allow pkg="org.apache.kafka.common.metrics" />
<allow pkg="org.apache.kafka.common.security" />
<allow class="org.apache.kafka.common.requests.ApiVersionsResponse" />

@ijuma ijuma Feb 6, 2021

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The network package is not meant to access requests, right? Could we make it work by changing the supplier to something more generic?

@hachikuji hachikuji Feb 6, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm.. Do you have any ideas? I guess I would say that the dependency is there regardless of how we choose to hide it since we have to go through ChannelBuilders to build the authenticator, and the authenticator does depend on the request APIs.

@cmccabe cmccabe Feb 18, 2021 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I guess I would say that the dependency is there regardless of how we choose to hide it since we have to go through ChannelBuilders to build the authenticator, and the authenticator does depend on the request APIs.

Yeah, the API versions response is pretty special in the protocol. Fortunately or unfortunately, it's not treated as a generic response. I tried to think of some ways to avoid this dependency but they all ended up being kind of like obfuscation.

Could we make it work by changing the supplier to something more generic?

Maybe I'm misinterpreting the suggestion, but if we choose to make this a Supplier<AbstractResponse>, that's still in the requests package....

@ijuma ijuma Feb 18, 2021 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I meant that you could introduce an interface that the response implements. That interface would live in the network package or in the common package.

@cmccabe cmccabe Feb 18, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm. Let's revisit this after 2.8.

</subpackage>

<subpackage name="resource">
Expand Down
10 changes: 5 additions & 5 deletions clients/src/main/java/org/apache/kafka/clients/NodeApiVersions.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,9 +16,6 @@
*/
package org.apache.kafka.clients;

import java.util.Collection;
import java.util.Collections;
import java.util.LinkedList;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersion;
import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersionCollection;
Expand All @@ -27,7 +24,10 @@
import org.apache.kafka.common.utils.Utils;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.EnumMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
Expand Down Expand Up @@ -62,7 +62,7 @@ public static NodeApiVersions create() {
*/
public static NodeApiVersions create(Collection<ApiVersion> overrides) {
List<ApiVersion> apiVersions = new LinkedList<>(overrides);
for (ApiKeys apiKey : ApiKeys.brokerApis()) {
for (ApiKeys apiKey : ApiKeys.zkBrokerApis()) {
boolean exists = false;
for (ApiVersion apiVersion : apiVersions) {
if (apiVersion.apiKey() == apiKey.id) {
Expand Down Expand Up @@ -170,7 +170,7 @@ public String toString(boolean lineBreaks) {

// Also handle the case where some apiKey types are not specified at all in the given ApiVersions,
// which may happen when the remote is too old.
for (ApiKeys apiKey : ApiKeys.brokerApis()) {
for (ApiKeys apiKey : ApiKeys.zkBrokerApis()) {
if (!apiKeysText.containsKey(apiKey.id)) {
StringBuilder bld = new StringBuilder();
bld.append(apiKey.name).append("(").
Expand Down
16 changes: 11 additions & 5 deletions clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import org.apache.kafka.common.config.SslClientAuth;
import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
import org.apache.kafka.common.errors.InvalidConfigurationException;
import org.apache.kafka.common.requests.ApiVersionsResponse;
import org.apache.kafka.common.security.auth.SecurityProtocol;
import org.apache.kafka.common.security.JaasContext;
import org.apache.kafka.common.security.authenticator.DefaultKafkaPrincipalBuilder;
Expand All @@ -40,6 +41,7 @@
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.function.Supplier;

public class ChannelBuilders {
private static final Logger log = LoggerFactory.getLogger(ChannelBuilders.class);
Expand Down Expand Up @@ -77,7 +79,7 @@ public static ChannelBuilder clientChannelBuilder(
throw new IllegalArgumentException("`clientSaslMechanism` must be non-null in client mode if `securityProtocol` is `" + securityProtocol + "`");
}
return create(securityProtocol, Mode.CLIENT, contextType, config, listenerName, false, clientSaslMechanism,
saslHandshakeRequestEnable, null, null, time, logContext);
saslHandshakeRequestEnable, null, null, time, logContext, null);
}

/**
Expand All @@ -89,6 +91,7 @@ public static ChannelBuilder clientChannelBuilder(
* @param tokenCache Delegation token cache
* @param time the time instance
* @param logContext the log context instance
* @param apiVersionSupplier supplier for ApiVersions responses sent prior to authentication
*
* @return the configured `ChannelBuilder`
*/
Expand All @@ -99,10 +102,11 @@ public static ChannelBuilder serverChannelBuilder(ListenerName listenerName,
CredentialCache credentialCache,
DelegationTokenCache tokenCache,
Time time,
LogContext logContext) {
LogContext logContext,
Supplier<ApiVersionsResponse> apiVersionSupplier) {
return create(securityProtocol, Mode.SERVER, JaasContext.Type.SERVER, config, listenerName,
isInterBrokerListener, null, true, credentialCache,
tokenCache, time, logContext);
tokenCache, time, logContext, apiVersionSupplier);
}

private static ChannelBuilder create(SecurityProtocol securityProtocol,
Expand All @@ -116,7 +120,8 @@ private static ChannelBuilder create(SecurityProtocol securityProtocol,
CredentialCache credentialCache,
DelegationTokenCache tokenCache,
Time time,
LogContext logContext) {
LogContext logContext,
Supplier<ApiVersionsResponse> apiVersionSupplier) {
Map<String, Object> configs = channelBuilderConfigs(config, listenerName);

ChannelBuilder channelBuilder;
Expand Down Expand Up @@ -174,7 +179,8 @@ private static ChannelBuilder create(SecurityProtocol securityProtocol,
tokenCache,
sslClientAuthOverride,
time,
logContext);
logContext,
apiVersionSupplier);
break;
case PLAINTEXT:
channelBuilder = new PlaintextChannelBuilder(listenerName);
Expand Down
12 changes: 10 additions & 2 deletions clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import org.apache.kafka.common.config.SslConfigs;
import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
import org.apache.kafka.common.memory.MemoryPool;
import org.apache.kafka.common.requests.ApiVersionsResponse;
import org.apache.kafka.common.security.JaasContext;
import org.apache.kafka.common.security.auth.AuthenticateCallbackHandler;
import org.apache.kafka.common.security.auth.Login;
Expand Down Expand Up @@ -85,6 +86,7 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl
private final DelegationTokenCache tokenCache;
private final Map<String, LoginManager> loginManagers;
private final Map<String, Subject> subjects;
private final Supplier<ApiVersionsResponse> apiVersionSupplier;

private SslFactory sslFactory;
private Map<String, ?> configs;
Expand All @@ -108,7 +110,8 @@ public SaslChannelBuilder(Mode mode,
DelegationTokenCache tokenCache,
String sslClientAuthOverride,
Time time,
LogContext logContext) {
LogContext logContext,
Supplier<ApiVersionsResponse> apiVersionSupplier) {
this.mode = mode;
this.jaasContexts = jaasContexts;
this.loginManagers = new HashMap<>(jaasContexts.size());
Expand All @@ -126,6 +129,11 @@ public SaslChannelBuilder(Mode mode,
this.time = time;
this.logContext = logContext;
this.log = logContext.logger(getClass());
this.apiVersionSupplier = apiVersionSupplier;

if (mode == Mode.SERVER && apiVersionSupplier == null) {
throw new IllegalArgumentException("Server channel builder must provide an ApiVersionResponse supplier");
}
}

@SuppressWarnings("unchecked")
Expand Down Expand Up @@ -266,7 +274,7 @@ protected SaslServerAuthenticator buildServerAuthenticator(Map<String, ?> config
ChannelMetadataRegistry metadataRegistry) {
return new SaslServerAuthenticator(configs, callbackHandlers, id, subjects,
kerberosShortNamer, listenerName, securityProtocol, transportLayer,
connectionsMaxReauthMsByMechanism, metadataRegistry, time);
connectionsMaxReauthMsByMechanism, metadataRegistry, time, apiVersionSupplier);
}

// Visible to override for testing
Expand Down
71 changes: 47 additions & 24 deletions clients/src/main/java/org/apache/kafka/common/protocol/ApiKeys.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,10 @@
import org.apache.kafka.common.protocol.types.Type;
import org.apache.kafka.common.record.RecordBatch;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.EnumMap;
import java.util.EnumSet;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicBoolean;
Expand Down Expand Up @@ -90,19 +93,28 @@ public enum ApiKeys {
ALTER_CLIENT_QUOTAS(ApiMessageType.ALTER_CLIENT_QUOTAS, false, true),
DESCRIBE_USER_SCRAM_CREDENTIALS(ApiMessageType.DESCRIBE_USER_SCRAM_CREDENTIALS),
ALTER_USER_SCRAM_CREDENTIALS(ApiMessageType.ALTER_USER_SCRAM_CREDENTIALS, false, true),
VOTE(ApiMessageType.VOTE, true, RecordBatch.MAGIC_VALUE_V0, false, true),
BEGIN_QUORUM_EPOCH(ApiMessageType.BEGIN_QUORUM_EPOCH, true, RecordBatch.MAGIC_VALUE_V0, false, true),
END_QUORUM_EPOCH(ApiMessageType.END_QUORUM_EPOCH, true, RecordBatch.MAGIC_VALUE_V0, false, true),
DESCRIBE_QUORUM(ApiMessageType.DESCRIBE_QUORUM, true, RecordBatch.MAGIC_VALUE_V0, false, true),
VOTE(ApiMessageType.VOTE, true, RecordBatch.MAGIC_VALUE_V0, false),
BEGIN_QUORUM_EPOCH(ApiMessageType.BEGIN_QUORUM_EPOCH, true, RecordBatch.MAGIC_VALUE_V0, false),
END_QUORUM_EPOCH(ApiMessageType.END_QUORUM_EPOCH, true, RecordBatch.MAGIC_VALUE_V0, false),
DESCRIBE_QUORUM(ApiMessageType.DESCRIBE_QUORUM, true, RecordBatch.MAGIC_VALUE_V0, false),
ALTER_ISR(ApiMessageType.ALTER_ISR, true),
UPDATE_FEATURES(ApiMessageType.UPDATE_FEATURES, false, true),
ENVELOPE(ApiMessageType.ENVELOPE, true, RecordBatch.MAGIC_VALUE_V0, false, true),
FETCH_SNAPSHOT(ApiMessageType.FETCH_SNAPSHOT, false, RecordBatch.MAGIC_VALUE_V0, false, true),
ENVELOPE(ApiMessageType.ENVELOPE, true, RecordBatch.MAGIC_VALUE_V0, false),
FETCH_SNAPSHOT(ApiMessageType.FETCH_SNAPSHOT, false, RecordBatch.MAGIC_VALUE_V0, false),
DESCRIBE_CLUSTER(ApiMessageType.DESCRIBE_CLUSTER),
DESCRIBE_PRODUCERS(ApiMessageType.DESCRIBE_PRODUCERS),
BROKER_REGISTRATION(ApiMessageType.BROKER_REGISTRATION, true, RecordBatch.MAGIC_VALUE_V0, false, true),
BROKER_HEARTBEAT(ApiMessageType.BROKER_HEARTBEAT, true, RecordBatch.MAGIC_VALUE_V0, false, true),
UNREGISTER_BROKER(ApiMessageType.UNREGISTER_BROKER, false, RecordBatch.MAGIC_VALUE_V0, true, false);
BROKER_REGISTRATION(ApiMessageType.BROKER_REGISTRATION, true, RecordBatch.MAGIC_VALUE_V0, false),
BROKER_HEARTBEAT(ApiMessageType.BROKER_HEARTBEAT, true, RecordBatch.MAGIC_VALUE_V0, false),
UNREGISTER_BROKER(ApiMessageType.UNREGISTER_BROKER, false, RecordBatch.MAGIC_VALUE_V0, true);

private static final Map<ApiMessageType.ListenerType, EnumSet<ApiKeys>> APIS_BY_LISTENER =
new EnumMap<>(ApiMessageType.ListenerType.class);

static {
for (ApiMessageType.ListenerType listenerType : ApiMessageType.ListenerType.values()) {
APIS_BY_LISTENER.put(listenerType, filterApisForListener(listenerType));
}
}

// The generator ensures every `ApiMessageType` has a unique id
private static final Map<Integer, ApiKeys> ID_TO_TYPE = Arrays.stream(ApiKeys.values())
Expand All @@ -120,9 +132,6 @@ public enum ApiKeys {
/** indicates the minimum required inter broker magic required to support the API */
public final byte minRequiredInterBrokerMagic;

/** indicates whether this is an API which is only exposed by the KIP-500 controller **/
public final boolean isControllerOnlyApi;

/** indicates whether the API is enabled for forwarding **/
public final boolean forwardable;

Expand All @@ -142,24 +151,17 @@ public enum ApiKeys {
this(messageType, clusterAction, RecordBatch.MAGIC_VALUE_V0, forwardable);
}

ApiKeys(ApiMessageType messageType, boolean clusterAction, byte minRequiredInterBrokerMagic, boolean forwardable) {
this(messageType, clusterAction, minRequiredInterBrokerMagic, forwardable, false);
}

ApiKeys(
ApiMessageType messageType,
boolean clusterAction,
byte minRequiredInterBrokerMagic,
boolean forwardable,
boolean isControllerOnlyApi
boolean forwardable
) {
this.messageType = messageType;
this.id = messageType.apiKey();
this.name = messageType.name;
this.clusterAction = clusterAction;
this.minRequiredInterBrokerMagic = minRequiredInterBrokerMagic;
this.isControllerOnlyApi = isControllerOnlyApi;

this.requiresDelayedAllocation = forwardable || shouldRetainsBufferReference(messageType.requestSchemas());
this.forwardable = forwardable;
}
Expand Down Expand Up @@ -195,6 +197,14 @@ public short oldestVersion() {
return messageType.lowestSupportedVersion();
}

public List<Short> allVersions() {
List<Short> versions = new ArrayList<>(latestVersion() - oldestVersion() + 1);
for (short version = oldestVersion(); version < latestVersion(); version++) {
versions.add(version);
}
return versions;
}

public boolean isVersionSupported(short apiVersion) {
return apiVersion >= oldestVersion() && apiVersion <= latestVersion();
}
Expand All @@ -207,14 +217,18 @@ public short responseHeaderVersion(short apiVersion) {
return messageType.responseHeaderVersion(apiVersion);
}

public boolean inScope(ApiMessageType.ListenerType listener) {
return messageType.listeners().contains(listener);
}

private static String toHtml() {
final StringBuilder b = new StringBuilder();
b.append("<table class=\"data-table\"><tbody>\n");
b.append("<tr>");
b.append("<th>Name</th>\n");
b.append("<th>Key</th>\n");
b.append("</tr>");
for (ApiKeys key : ApiKeys.brokerApis()) {
for (ApiKeys key : zkBrokerApis()) {
b.append("<tr>\n");
b.append("<td>");
b.append("<a href=\"#The_Messages_" + key.name + "\">" + key.name + "</a>");
Expand Down Expand Up @@ -246,10 +260,19 @@ public void visit(Type field) {
return hasBuffer.get();
}

public static List<ApiKeys> brokerApis() {
return Arrays.stream(values())
.filter(api -> !api.isControllerOnlyApi)
public static EnumSet<ApiKeys> zkBrokerApis() {
return apisForListener(ApiMessageType.ListenerType.ZK_BROKER);
}

public static EnumSet<ApiKeys> apisForListener(ApiMessageType.ListenerType listener) {
Comment thread
hachikuji marked this conversation as resolved.
return APIS_BY_LISTENER.get(listener);
}

private static EnumSet<ApiKeys> filterApisForListener(ApiMessageType.ListenerType listener) {
List<ApiKeys> controllerApis = Arrays.stream(ApiKeys.values())
.filter(apiKey -> apiKey.messageType.listeners().contains(listener))
.collect(Collectors.toList());
return EnumSet.copyOf(controllerApis);
}

}
2 changes: 1 addition & 1 deletion clients/src/main/java/org/apache/kafka/common/protocol/Protocol.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -133,7 +133,7 @@ public static String toHtml() {
b.append("</pre>\n");
schemaToFieldTableHtml(ResponseHeaderData.SCHEMAS[i], b);
}
for (ApiKeys key : ApiKeys.brokerApis()) {
for (ApiKeys key : ApiKeys.zkBrokerApis()) {
// Key
b.append("<h5>");
b.append("<a name=\"The_Messages_" + key.name + "\">");
Expand Down
Loading