You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2024/07/02 18:46:11 UTC
(kafka) branch trunk updated: KAFKA-3346 Rename Mode to ConnectionMode (#16403)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 6897b06b039 KAFKA-3346 Rename Mode to ConnectionMode (#16403)
6897b06b039 is described below
commit 6897b06b03917847f599c9e30897228233eee78a
Author: abhi-ksolves <11...@users.noreply.github.com>
AuthorDate: Wed Jul 3 00:16:04 2024 +0530
KAFKA-3346 Rename Mode to ConnectionMode (#16403)
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
.../kafka/common/network/ChannelBuilders.java | 20 ++---
.../network/{Mode.java => ConnectionMode.java} | 2 +-
.../kafka/common/network/SaslChannelBuilder.java | 18 ++--
.../kafka/common/network/SslChannelBuilder.java | 8 +-
.../internals/secured/JaasOptionsUtils.java | 4 +-
.../security/ssl/DefaultSslEngineFactory.java | 10 +--
.../kafka/common/security/ssl/SslFactory.java | 30 +++----
.../apache/kafka/common/network/CertStores.java | 4 +-
.../apache/kafka/common/network/EchoServer.java | 2 +-
.../common/network/SaslChannelBuilderTest.java | 4 +-
.../kafka/common/network/SslSelectorTest.java | 20 ++---
.../common/network/SslTransportLayerTest.java | 20 ++---
.../common/network/SslTransportTls12Tls13Test.java | 4 +-
.../network/SslVersionsTransportLayerTest.java | 2 +-
.../kafka/common/network/Tls12SelectorTest.java | 2 +-
.../kafka/common/network/Tls13SelectorTest.java | 2 +-
.../authenticator/SaslAuthenticatorTest.java | 14 ++--
.../kafka/common/security/ssl/SslFactoryTest.java | 96 +++++++++++-----------
.../java/org/apache/kafka/test/TestSslUtils.java | 34 ++++----
.../MirrorConnectorsIntegrationSSLTest.java | 4 +-
.../integration/RestForwardingIntegrationTest.java | 4 +-
.../kafka/api/BaseProducerSendTest.scala | 4 +-
.../kafka/api/IntegrationTestHarness.scala | 4 +-
.../SaslPlainSslEndToEndAuthorizationTest.scala | 4 +-
.../kafka/api/SslAdminIntegrationTest.scala | 6 +-
.../kafka/api/SslEndToEndAuthorizationTest.scala | 6 +-
.../server/DynamicBrokerReconfigurationTest.scala | 6 +-
.../kafka/server/GssapiAuthenticationTest.scala | 2 +-
.../kafka/server/IntegrationTestUtils.scala | 4 +-
...ListenersWithSameSecurityProtocolBaseTest.scala | 4 +-
.../test/scala/unit/kafka/utils/TestUtils.scala | 18 ++--
.../integration/ResetIntegrationWithSslTest.java | 4 +-
32 files changed, 183 insertions(+), 183 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java b/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java
index f393db85c93..292ec4c41ad 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/ChannelBuilders.java
@@ -79,7 +79,7 @@ public class ChannelBuilders {
if (clientSaslMechanism == null)
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,
+ return create(securityProtocol, ConnectionMode.CLIENT, contextType, config, listenerName, false, clientSaslMechanism,
saslHandshakeRequestEnable, null, null, time, logContext, null);
}
@@ -105,13 +105,13 @@ public class ChannelBuilders {
Time time,
LogContext logContext,
Supplier<ApiVersionsResponse> apiVersionSupplier) {
- return create(securityProtocol, Mode.SERVER, JaasContext.Type.SERVER, config, listenerName,
+ return create(securityProtocol, ConnectionMode.SERVER, JaasContext.Type.SERVER, config, listenerName,
isInterBrokerListener, null, true, credentialCache,
tokenCache, time, logContext, apiVersionSupplier);
}
private static ChannelBuilder create(SecurityProtocol securityProtocol,
- Mode mode,
+ ConnectionMode connectionMode,
JaasContext.Type contextType,
AbstractConfig config,
ListenerName listenerName,
@@ -128,15 +128,15 @@ public class ChannelBuilders {
ChannelBuilder channelBuilder;
switch (securityProtocol) {
case SSL:
- requireNonNullMode(mode, securityProtocol);
- channelBuilder = new SslChannelBuilder(mode, listenerName, isInterBrokerListener, logContext);
+ requireNonNullMode(connectionMode, securityProtocol);
+ channelBuilder = new SslChannelBuilder(connectionMode, listenerName, isInterBrokerListener, logContext);
break;
case SASL_SSL:
case SASL_PLAINTEXT:
- requireNonNullMode(mode, securityProtocol);
+ requireNonNullMode(connectionMode, securityProtocol);
Map<String, JaasContext> jaasContexts;
String sslClientAuthOverride = null;
- if (mode == Mode.SERVER) {
+ if (connectionMode == ConnectionMode.SERVER) {
@SuppressWarnings("unchecked")
List<String> enabledMechanisms = (List<String>) configs.get(BrokerSecurityConfigs.SASL_ENABLED_MECHANISMS_CONFIG);
jaasContexts = new HashMap<>(enabledMechanisms.size());
@@ -169,7 +169,7 @@ public class ChannelBuilders {
JaasContext.loadServerContext(listenerName, clientSaslMechanism, configs);
jaasContexts = Collections.singletonMap(clientSaslMechanism, jaasContext);
}
- channelBuilder = new SaslChannelBuilder(mode,
+ channelBuilder = new SaslChannelBuilder(connectionMode,
jaasContexts,
securityProtocol,
listenerName,
@@ -216,8 +216,8 @@ public class ChannelBuilders {
return parsedConfigs;
}
- private static void requireNonNullMode(Mode mode, SecurityProtocol securityProtocol) {
- if (mode == null)
+ private static void requireNonNullMode(ConnectionMode connectionMode, SecurityProtocol securityProtocol) {
+ if (connectionMode == null)
throw new IllegalArgumentException("`mode` must be non-null if `securityProtocol` is `" + securityProtocol + "`");
}
diff --git a/clients/src/main/java/org/apache/kafka/common/network/Mode.java b/clients/src/main/java/org/apache/kafka/common/network/ConnectionMode.java
similarity index 95%
rename from clients/src/main/java/org/apache/kafka/common/network/Mode.java
rename to clients/src/main/java/org/apache/kafka/common/network/ConnectionMode.java
index 61239703249..3c62cbd8ecc 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/Mode.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/ConnectionMode.java
@@ -19,4 +19,4 @@ package org.apache.kafka.common.network;
/**
* Connection mode for SSL and SASL connections.
*/
-public enum Mode { CLIENT, SERVER }
+public enum ConnectionMode { CLIENT, SERVER }
diff --git a/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java b/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java
index 8ed333d24d5..295d7dbebab 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/SaslChannelBuilder.java
@@ -81,7 +81,7 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl
private final ListenerName listenerName;
private final boolean isInterBrokerListener;
private final String clientSaslMechanism;
- private final Mode mode;
+ private final ConnectionMode connectionMode;
private final Map<String, JaasContext> jaasContexts;
private final boolean handshakeRequestEnable;
private final CredentialCache credentialCache;
@@ -100,7 +100,7 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl
private Map<String, ?> configs;
private KerberosShortNamer kerberosShortNamer;
- public SaslChannelBuilder(Mode mode,
+ public SaslChannelBuilder(ConnectionMode connectionMode,
Map<String, JaasContext> jaasContexts,
SecurityProtocol securityProtocol,
ListenerName listenerName,
@@ -113,7 +113,7 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl
Time time,
LogContext logContext,
Supplier<ApiVersionsResponse> apiVersionSupplier) {
- this.mode = mode;
+ this.connectionMode = connectionMode;
this.jaasContexts = jaasContexts;
this.loginManagers = new HashMap<>(jaasContexts.size());
this.subjects = new HashMap<>(jaasContexts.size());
@@ -132,7 +132,7 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl
this.log = logContext.logger(getClass());
this.apiVersionSupplier = apiVersionSupplier;
- if (mode == Mode.SERVER && apiVersionSupplier == null) {
+ if (connectionMode == ConnectionMode.SERVER && apiVersionSupplier == null) {
throw new IllegalArgumentException("Server channel builder must provide an ApiVersionResponse supplier");
}
}
@@ -142,7 +142,7 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl
public void configure(Map<String, ?> configs) throws KafkaException {
try {
this.configs = configs;
- if (mode == Mode.SERVER) {
+ if (connectionMode == ConnectionMode.SERVER) {
createServerCallbackHandlers(configs);
createConnectionsMaxReauthMsMap(configs);
} else
@@ -153,7 +153,7 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl
}
Class<? extends Login> defaultLoginClass = defaultLoginClass();
- if (mode == Mode.SERVER && jaasContexts.containsKey(SaslConfigs.GSSAPI_MECHANISM)) {
+ if (connectionMode == ConnectionMode.SERVER && jaasContexts.containsKey(SaslConfigs.GSSAPI_MECHANISM)) {
String defaultRealm;
try {
defaultRealm = defaultKerberosRealm();
@@ -172,12 +172,12 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl
loginManagers.put(mechanism, loginManager);
Subject subject = loginManager.subject();
subjects.put(mechanism, subject);
- if (mode == Mode.SERVER && mechanism.equals(SaslConfigs.GSSAPI_MECHANISM))
+ if (connectionMode == ConnectionMode.SERVER && mechanism.equals(SaslConfigs.GSSAPI_MECHANISM))
maybeAddNativeGssapiCredentials(subject);
}
if (this.securityProtocol == SecurityProtocol.SASL_SSL) {
// Disable SSL client authentication as we are using SASL authentication
- this.sslFactory = new SslFactory(mode, sslClientAuthOverride, isInterBrokerListener);
+ this.sslFactory = new SslFactory(connectionMode, sslClientAuthOverride, isInterBrokerListener);
this.sslFactory.configure(configs);
}
} catch (Throwable e) {
@@ -218,7 +218,7 @@ public class SaslChannelBuilder implements ChannelBuilder, ListenerReconfigurabl
transportLayer = buildTransportLayer(id, key, socketChannel, metadataRegistry);
final TransportLayer finalTransportLayer = transportLayer;
Supplier<Authenticator> authenticatorCreator;
- if (mode == Mode.SERVER) {
+ if (connectionMode == ConnectionMode.SERVER) {
authenticatorCreator = () -> buildServerAuthenticator(configs,
Collections.unmodifiableMap(saslCallbackHandlers),
id,
diff --git a/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java b/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java
index f5002289f14..b45fb07442e 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/SslChannelBuilder.java
@@ -42,7 +42,7 @@ import java.util.function.Supplier;
public class SslChannelBuilder implements ChannelBuilder, ListenerReconfigurable {
private final ListenerName listenerName;
private final boolean isInterBrokerListener;
- private final Mode mode;
+ private final ConnectionMode connectionMode;
private SslFactory sslFactory;
private Map<String, ?> configs;
private SslPrincipalMapper sslPrincipalMapper;
@@ -51,11 +51,11 @@ public class SslChannelBuilder implements ChannelBuilder, ListenerReconfigurable
* Constructs an SSL channel builder. ListenerName is provided only
* for server channel builder and will be null for client channel builder.
*/
- public SslChannelBuilder(Mode mode,
+ public SslChannelBuilder(ConnectionMode connectionMode,
ListenerName listenerName,
boolean isInterBrokerListener,
LogContext logContext) {
- this.mode = mode;
+ this.connectionMode = connectionMode;
this.listenerName = listenerName;
this.isInterBrokerListener = isInterBrokerListener;
}
@@ -66,7 +66,7 @@ public class SslChannelBuilder implements ChannelBuilder, ListenerReconfigurable
String sslPrincipalMappingRules = (String) configs.get(BrokerSecurityConfigs.SSL_PRINCIPAL_MAPPING_RULES_CONFIG);
if (sslPrincipalMappingRules != null)
sslPrincipalMapper = SslPrincipalMapper.fromRules(sslPrincipalMappingRules);
- this.sslFactory = new SslFactory(mode, null, isInterBrokerListener);
+ this.sslFactory = new SslFactory(connectionMode, null, isInterBrokerListener);
this.sslFactory.configure(this.configs);
} catch (KafkaException e) {
throw e;
diff --git a/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/JaasOptionsUtils.java b/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/JaasOptionsUtils.java
index 3336922d7b0..3e49595dbc1 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/JaasOptionsUtils.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/oauthbearer/internals/secured/JaasOptionsUtils.java
@@ -20,7 +20,7 @@ package org.apache.kafka.common.security.oauthbearer.internals.secured;
import org.apache.kafka.common.config.AbstractConfig;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.common.config.ConfigException;
-import org.apache.kafka.common.network.Mode;
+import org.apache.kafka.common.network.ConnectionMode;
import org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule;
import org.apache.kafka.common.security.ssl.DefaultSslEngineFactory;
import org.apache.kafka.common.security.ssl.SslFactory;
@@ -75,7 +75,7 @@ public class JaasOptionsUtils {
public SSLSocketFactory createSSLSocketFactory() {
Map<String, ?> sslClientConfig = getSslClientConfig();
- SslFactory sslFactory = new SslFactory(Mode.CLIENT);
+ SslFactory sslFactory = new SslFactory(ConnectionMode.CLIENT);
sslFactory.configure(sslClientConfig);
SSLSocketFactory socketFactory = ((DefaultSslEngineFactory) sslFactory.sslEngineFactory()).sslContext().getSocketFactory();
log.debug("Created SSLSocketFactory: {}", sslClientConfig);
diff --git a/clients/src/main/java/org/apache/kafka/common/security/ssl/DefaultSslEngineFactory.java b/clients/src/main/java/org/apache/kafka/common/security/ssl/DefaultSslEngineFactory.java
index de23000215e..0a3d587df90 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/ssl/DefaultSslEngineFactory.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/ssl/DefaultSslEngineFactory.java
@@ -22,7 +22,7 @@ import org.apache.kafka.common.config.SslConfigs;
import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
import org.apache.kafka.common.config.types.Password;
import org.apache.kafka.common.errors.InvalidConfigurationException;
-import org.apache.kafka.common.network.Mode;
+import org.apache.kafka.common.network.ConnectionMode;
import org.apache.kafka.common.security.auth.SslEngineFactory;
import org.apache.kafka.common.utils.SecurityUtils;
import org.apache.kafka.common.utils.Utils;
@@ -94,12 +94,12 @@ public class DefaultSslEngineFactory implements SslEngineFactory {
@Override
public SSLEngine createClientSslEngine(String peerHost, int peerPort, String endpointIdentification) {
- return createSslEngine(Mode.CLIENT, peerHost, peerPort, endpointIdentification);
+ return createSslEngine(ConnectionMode.CLIENT, peerHost, peerPort, endpointIdentification);
}
@Override
public SSLEngine createServerSslEngine(String peerHost, int peerPort) {
- return createSslEngine(Mode.SERVER, peerHost, peerPort, null);
+ return createSslEngine(ConnectionMode.SERVER, peerHost, peerPort, null);
}
@Override
@@ -184,12 +184,12 @@ public class DefaultSslEngineFactory implements SslEngineFactory {
return this.sslContext;
}
- private SSLEngine createSslEngine(Mode mode, String peerHost, int peerPort, String endpointIdentification) {
+ private SSLEngine createSslEngine(ConnectionMode connectionMode, String peerHost, int peerPort, String endpointIdentification) {
SSLEngine sslEngine = sslContext.createSSLEngine(peerHost, peerPort);
if (cipherSuites != null) sslEngine.setEnabledCipherSuites(cipherSuites);
if (enabledProtocols != null) sslEngine.setEnabledProtocols(enabledProtocols);
- if (mode == Mode.SERVER) {
+ if (connectionMode == ConnectionMode.SERVER) {
sslEngine.setUseClientMode(false);
switch (sslClientAuth) {
case REQUIRED:
diff --git a/clients/src/main/java/org/apache/kafka/common/security/ssl/SslFactory.java b/clients/src/main/java/org/apache/kafka/common/security/ssl/SslFactory.java
index 7113a48d969..361e1c23439 100644
--- a/clients/src/main/java/org/apache/kafka/common/security/ssl/SslFactory.java
+++ b/clients/src/main/java/org/apache/kafka/common/security/ssl/SslFactory.java
@@ -21,7 +21,7 @@ import org.apache.kafka.common.Reconfigurable;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.config.SslConfigs;
import org.apache.kafka.common.config.internals.BrokerSecurityConfigs;
-import org.apache.kafka.common.network.Mode;
+import org.apache.kafka.common.network.ConnectionMode;
import org.apache.kafka.common.security.auth.SslEngineFactory;
import org.apache.kafka.common.utils.ConfigUtils;
import org.apache.kafka.common.utils.Utils;
@@ -56,30 +56,30 @@ import javax.net.ssl.SSLException;
public class SslFactory implements Reconfigurable, Closeable {
private static final Logger log = LoggerFactory.getLogger(SslFactory.class);
- private final Mode mode;
+ private final ConnectionMode connectionMode;
private final String clientAuthConfigOverride;
private final boolean keystoreVerifiableUsingTruststore;
private String endpointIdentification;
private SslEngineFactory sslEngineFactory;
private Map<String, Object> sslEngineFactoryConfig;
- public SslFactory(Mode mode) {
- this(mode, null, false);
+ public SslFactory(ConnectionMode connectionMode) {
+ this(connectionMode, null, false);
}
/**
* Create an SslFactory.
*
- * @param mode Whether to use client or server mode.
+ * @param connectionMode Whether to use client or server mode.
* @param clientAuthConfigOverride The value to override ssl.client.auth with, or null
* if we don't want to override it.
* @param keystoreVerifiableUsingTruststore True if we should require the keystore to be verifiable
* using the truststore.
*/
- public SslFactory(Mode mode,
+ public SslFactory(ConnectionMode connectionMode,
String clientAuthConfigOverride,
boolean keystoreVerifiableUsingTruststore) {
- this.mode = mode;
+ this.connectionMode = connectionMode;
this.clientAuthConfigOverride = clientAuthConfigOverride;
this.keystoreVerifiableUsingTruststore = keystoreVerifiableUsingTruststore;
}
@@ -125,7 +125,7 @@ public class SslFactory implements Reconfigurable, Closeable {
if (newSslEngineFactory != this.sslEngineFactory) {
Utils.closeQuietly(this.sslEngineFactory, "close stale ssl engine factory");
this.sslEngineFactory = newSslEngineFactory;
- log.info("Created new {} SSL engine builder with keystore {} truststore {}", mode,
+ log.info("Created new {} SSL engine builder with keystore {} truststore {}", connectionMode,
newSslEngineFactory.keystore(), newSslEngineFactory.truststore());
}
}
@@ -203,7 +203,7 @@ public class SslFactory implements Reconfigurable, Closeable {
if (sslEngineFactory == null) {
throw new IllegalStateException("SslFactory has not been configured.");
}
- if (mode == Mode.SERVER) {
+ if (connectionMode == ConnectionMode.SERVER) {
return sslEngineFactory.createServerSslEngine(peerHost, peerPort);
} else {
return sslEngineFactory.createClientSslEngine(peerHost, peerPort, endpointIdentification);
@@ -406,15 +406,15 @@ public class SslFactory implements Reconfigurable, Closeable {
static void validate(SslEngineFactory oldEngineBuilder,
SslEngineFactory newEngineBuilder) throws SSLException {
- validate(createSslEngineForValidation(oldEngineBuilder, Mode.SERVER),
- createSslEngineForValidation(newEngineBuilder, Mode.CLIENT));
- validate(createSslEngineForValidation(newEngineBuilder, Mode.SERVER),
- createSslEngineForValidation(oldEngineBuilder, Mode.CLIENT));
+ validate(createSslEngineForValidation(oldEngineBuilder, ConnectionMode.SERVER),
+ createSslEngineForValidation(newEngineBuilder, ConnectionMode.CLIENT));
+ validate(createSslEngineForValidation(newEngineBuilder, ConnectionMode.SERVER),
+ createSslEngineForValidation(oldEngineBuilder, ConnectionMode.CLIENT));
}
- private static SSLEngine createSslEngineForValidation(SslEngineFactory sslEngineFactory, Mode mode) {
+ private static SSLEngine createSslEngineForValidation(SslEngineFactory sslEngineFactory, ConnectionMode connectionMode) {
// Use empty hostname, disable hostname verification
- if (mode == Mode.SERVER) {
+ if (connectionMode == ConnectionMode.SERVER) {
return sslEngineFactory.createServerSslEngine("", 0);
} else {
return sslEngineFactory.createClientSslEngine("", 0, "");
diff --git a/clients/src/test/java/org/apache/kafka/common/network/CertStores.java b/clients/src/test/java/org/apache/kafka/common/network/CertStores.java
index 96859d1cea5..47d8b86a746 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/CertStores.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/CertStores.java
@@ -62,9 +62,9 @@ public class CertStores {
private CertStores(boolean server, String commonName, String keyAlgorithm, TestSslUtils.CertificateBuilder certBuilder, boolean usePem) throws Exception {
String name = server ? "server" : "client";
- Mode mode = server ? Mode.SERVER : Mode.CLIENT;
+ ConnectionMode connectionMode = server ? ConnectionMode.SERVER : ConnectionMode.CLIENT;
File truststoreFile = usePem ? null : TestUtils.tempFile(name + "TS", ".jks");
- sslConfig = new SslConfigsBuilder(mode)
+ sslConfig = new SslConfigsBuilder(connectionMode)
.useClientCert(!server)
.certAlias(name)
.cn(commonName)
diff --git a/clients/src/test/java/org/apache/kafka/common/network/EchoServer.java b/clients/src/test/java/org/apache/kafka/common/network/EchoServer.java
index bfe8393e695..8ab50f6465f 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/EchoServer.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/EchoServer.java
@@ -57,7 +57,7 @@ class EchoServer extends Thread {
public EchoServer(SecurityProtocol securityProtocol, Map<String, ?> configs) throws Exception {
switch (securityProtocol) {
case SSL:
- this.sslFactory = new SslFactory(Mode.SERVER);
+ this.sslFactory = new SslFactory(ConnectionMode.SERVER);
this.sslFactory.configure(configs);
SSLContext sslContext = ((DefaultSslEngineFactory) this.sslFactory.sslEngineFactory()).sslContext();
this.serverSocket = sslContext.getServerSocketFactory().createServerSocket(0);
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SaslChannelBuilderTest.java b/clients/src/test/java/org/apache/kafka/common/network/SaslChannelBuilderTest.java
index 2169d7ac73b..f8ba8dc05b3 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SaslChannelBuilderTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SaslChannelBuilderTest.java
@@ -165,7 +165,7 @@ public class SaslChannelBuilderTest {
}
private SaslChannelBuilder createGssapiChannelBuilder(Map<String, JaasContext> jaasContexts, GSSManager gssManager) {
- SaslChannelBuilder channelBuilder = new SaslChannelBuilder(Mode.SERVER, jaasContexts,
+ SaslChannelBuilder channelBuilder = new SaslChannelBuilder(ConnectionMode.SERVER, jaasContexts,
SecurityProtocol.SASL_PLAINTEXT, new ListenerName("GSSAPI"), false, "GSSAPI",
true, null, null, null, Time.SYSTEM, new LogContext(), defaultApiVersionsSupplier()) {
@@ -205,7 +205,7 @@ public class SaslChannelBuilderTest {
jaasConfig.addEntry("jaasContext", loginModule.getName(), new HashMap<>());
JaasContext jaasContext = new JaasContext("jaasContext", JaasContext.Type.SERVER, jaasConfig, null);
Map<String, JaasContext> jaasContexts = Collections.singletonMap(saslMechanism, jaasContext);
- return new SaslChannelBuilder(Mode.CLIENT, jaasContexts, securityProtocol, new ListenerName(saslMechanism),
+ return new SaslChannelBuilder(ConnectionMode.CLIENT, jaasContexts, securityProtocol, new ListenerName(saslMechanism),
false, saslMechanism, true, null,
null, null, Time.SYSTEM, new LogContext(), defaultApiVersionsSupplier());
}
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java b/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java
index bb31d5cf44b..8a9704c1621 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SslSelectorTest.java
@@ -69,13 +69,13 @@ public abstract class SslSelectorTest extends SelectorTest {
public void setUp() throws Exception {
File trustStoreFile = TestUtils.tempFile("truststore", ".jks");
- Map<String, Object> sslServerConfigs = TestSslUtils.createSslConfig(false, true, Mode.SERVER, trustStoreFile, "server");
+ Map<String, Object> sslServerConfigs = TestSslUtils.createSslConfig(false, true, ConnectionMode.SERVER, trustStoreFile, "server");
this.server = new EchoServer(SecurityProtocol.SSL, sslServerConfigs);
this.server.start();
this.time = new MockTime();
sslClientConfigs = createSslClientConfigs(trustStoreFile);
LogContext logContext = new LogContext();
- this.channelBuilder = new SslChannelBuilder(Mode.CLIENT, null, false, logContext);
+ this.channelBuilder = new SslChannelBuilder(ConnectionMode.CLIENT, null, false, logContext);
this.channelBuilder.configure(sslClientConfigs);
this.metrics = new Metrics();
this.selector = new Selector(5000, metrics, time, "MetricGroup", channelBuilder, logContext);
@@ -113,9 +113,9 @@ public abstract class SslSelectorTest extends SelectorTest {
server.start();
Time time = new MockTime();
File trustStoreFile = new File(TestKeyManagerFactory.TestKeyManager.mockTrustStoreFile);
- Map<String, Object> sslClientConfigs = TestSslUtils.createSslConfig(true, true, Mode.CLIENT, trustStoreFile, "client");
+ Map<String, Object> sslClientConfigs = TestSslUtils.createSslConfig(true, true, ConnectionMode.CLIENT, trustStoreFile, "client");
- ChannelBuilder channelBuilder = new TestSslChannelBuilder(Mode.CLIENT);
+ ChannelBuilder channelBuilder = new TestSslChannelBuilder(ConnectionMode.CLIENT);
channelBuilder.configure(sslClientConfigs);
Metrics metrics = new Metrics();
Selector selector = new Selector(5000, metrics, time, "MetricGroup", channelBuilder, new LogContext());
@@ -152,7 +152,7 @@ public abstract class SslSelectorTest extends SelectorTest {
this.selector.close();
- this.channelBuilder = new TestSslChannelBuilder(Mode.CLIENT);
+ this.channelBuilder = new TestSslChannelBuilder(ConnectionMode.CLIENT);
this.channelBuilder.configure(sslClientConfigs);
this.selector = new Selector(5000, metrics, time, "MetricGroup", channelBuilder, new LogContext());
connect(node, new InetSocketAddress("localhost", server.port));
@@ -194,7 +194,7 @@ public abstract class SslSelectorTest extends SelectorTest {
String node2 = "2";
final AtomicInteger node1Polls = new AtomicInteger();
- this.channelBuilder = new TestSslChannelBuilder(Mode.CLIENT);
+ this.channelBuilder = new TestSslChannelBuilder(ConnectionMode.CLIENT);
this.channelBuilder.configure(sslClientConfigs);
this.selector = new Selector(5000, metrics, time, "MetricGroup", channelBuilder, new LogContext()) {
@Override
@@ -251,11 +251,11 @@ public abstract class SslSelectorTest extends SelectorTest {
//the initial channel builder is for clients, we need a server one
String tlsProtocol = "TLSv1.2";
File trustStoreFile = TestUtils.tempFile("truststore", ".jks");
- Map<String, Object> sslServerConfigs = new TestSslUtils.SslConfigsBuilder(Mode.SERVER)
+ Map<String, Object> sslServerConfigs = new TestSslUtils.SslConfigsBuilder(ConnectionMode.SERVER)
.tlsProtocol(tlsProtocol)
.createNewTrustStore(trustStoreFile)
.build();
- channelBuilder = new SslChannelBuilder(Mode.SERVER, null, false, new LogContext());
+ channelBuilder = new SslChannelBuilder(ConnectionMode.SERVER, null, false, new LogContext());
channelBuilder.configure(sslServerConfigs);
selector = new Selector(NetworkReceive.UNLIMITED, 5000, metrics, time, "MetricGroup",
new HashMap<>(), true, false, channelBuilder, pool, new LogContext());
@@ -341,8 +341,8 @@ public abstract class SslSelectorTest extends SelectorTest {
private static class TestSslChannelBuilder extends SslChannelBuilder {
- public TestSslChannelBuilder(Mode mode) {
- super(mode, null, false, new LogContext());
+ public TestSslChannelBuilder(ConnectionMode connectionMode) {
+ super(connectionMode, null, false, new LogContext());
}
@Override
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java b/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java
index 0ed557ba744..c667fe5d453 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java
@@ -264,7 +264,7 @@ public class SslTransportLayerTest {
args.sslClientConfigs = args.getTrustingConfig(args.clientCertStores, args.serverCertStores);
// Create a server with endpoint validation enabled on the server SSL engine
- SslChannelBuilder serverChannelBuilder = new TestSslChannelBuilder(Mode.SERVER) {
+ SslChannelBuilder serverChannelBuilder = new TestSslChannelBuilder(ConnectionMode.SERVER) {
@Override
protected TestSslTransportLayer newTransportLayer(String id, SelectionKey key, SSLEngine sslEngine) throws IOException {
SSLParameters sslParams = sslEngine.getSSLParameters();
@@ -764,7 +764,7 @@ public class SslTransportLayerTest {
@ArgumentsSource(SslTransportLayerArgumentsProvider.class)
public void testNetworkThreadTimeRecorded(Args args) throws Exception {
LogContext logContext = new LogContext();
- ChannelBuilder channelBuilder = new SslChannelBuilder(Mode.CLIENT, null, false, logContext);
+ ChannelBuilder channelBuilder = new SslChannelBuilder(ConnectionMode.CLIENT, null, false, logContext);
channelBuilder.configure(args.sslClientConfigs);
try (Selector selector = new Selector(NetworkReceive.UNLIMITED, Selector.NO_IDLE_TIMEOUT_MS, new Metrics(), Time.SYSTEM,
"MetricGroup", new HashMap<>(), false, true, channelBuilder, MemoryPool.NONE, logContext)) {
@@ -874,7 +874,7 @@ public class SslTransportLayerTest {
private void testIOExceptionsDuringHandshake(Args args,
FailureAction readFailureAction,
FailureAction flushFailureAction) throws Exception {
- TestSslChannelBuilder channelBuilder = new TestSslChannelBuilder(Mode.CLIENT);
+ TestSslChannelBuilder channelBuilder = new TestSslChannelBuilder(ConnectionMode.CLIENT);
boolean done = false;
for (int i = 1; i <= 100; i++) {
String node = String.valueOf(i);
@@ -924,7 +924,7 @@ public class SslTransportLayerTest {
// Test without delay and a couple of delay counts to ensure delay applies to handshake failure
for (int i = 0; i < 3; i++) {
String node = String.valueOf(i);
- TestSslChannelBuilder serverChannelBuilder = new TestSslChannelBuilder(Mode.SERVER);
+ TestSslChannelBuilder serverChannelBuilder = new TestSslChannelBuilder(ConnectionMode.SERVER);
serverChannelBuilder.configure(args.sslServerConfigs);
serverChannelBuilder.flushDelayCount = i;
server = new NioEchoServer(ListenerName.forSecurityProtocol(SecurityProtocol.SSL),
@@ -963,7 +963,7 @@ public class SslTransportLayerTest {
}
private SslChannelBuilder newClientChannelBuilder() {
- return new SslChannelBuilder(Mode.CLIENT, null, false, new LogContext());
+ return new SslChannelBuilder(ConnectionMode.CLIENT, null, false, new LogContext());
}
private void testClose(Args args, SecurityProtocol securityProtocol, ChannelBuilder clientChannelBuilder) throws Exception {
@@ -1132,7 +1132,7 @@ public class SslTransportLayerTest {
TestSslUtils.CertificateBuilder certBuilder = new TestSslUtils.CertificateBuilder().sanDnsNames("localhost", "*.example.com");
String truststorePath = (String) args.sslClientConfigs.get(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG);
File truststoreFile = truststorePath != null ? new File(truststorePath) : null;
- TestSslUtils.SslConfigsBuilder builder = new TestSslUtils.SslConfigsBuilder(Mode.SERVER)
+ TestSslUtils.SslConfigsBuilder builder = new TestSslUtils.SslConfigsBuilder(ConnectionMode.SERVER)
.useClientCert(false)
.certAlias("server")
.cn("server")
@@ -1288,7 +1288,7 @@ public class SslTransportLayerTest {
private Selector createSelector(Map<String, Object> sslClientConfigs, final Integer netReadBufSize,
final Integer netWriteBufSize, final Integer appBufSize) {
- TestSslChannelBuilder channelBuilder = new TestSslChannelBuilder(Mode.CLIENT);
+ TestSslChannelBuilder channelBuilder = new TestSslChannelBuilder(ConnectionMode.CLIENT);
channelBuilder.configureBufferSizes(netReadBufSize, netWriteBufSize, appBufSize);
channelBuilder.configure(sslClientConfigs);
if (this.selector != null) {
@@ -1308,7 +1308,7 @@ public class SslTransportLayerTest {
private Selector createSelector(Args args) {
LogContext logContext = new LogContext();
- ChannelBuilder channelBuilder = new SslChannelBuilder(Mode.CLIENT, null, false, logContext);
+ ChannelBuilder channelBuilder = new SslChannelBuilder(ConnectionMode.CLIENT, null, false, logContext);
channelBuilder.configure(args.sslClientConfigs);
selector = new Selector(5000, new Metrics(), TIME, "MetricGroup", channelBuilder, logContext);
return selector;
@@ -1366,8 +1366,8 @@ public class SslTransportLayerTest {
FailureAction flushFailureAction = FailureAction.NO_OP;
int flushDelayCount = 0;
- public TestSslChannelBuilder(Mode mode) {
- super(mode, null, false, new LogContext());
+ public TestSslChannelBuilder(ConnectionMode connectionMode) {
+ super(connectionMode, null, false, new LogContext());
}
public void configureBufferSizes(Integer netReadBufSize, Integer netWriteBufSize, Integer appBufSize) {
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SslTransportTls12Tls13Test.java b/clients/src/test/java/org/apache/kafka/common/network/SslTransportTls12Tls13Test.java
index 8977872e60f..f5cac47ed9b 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SslTransportTls12Tls13Test.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SslTransportTls12Tls13Test.java
@@ -53,7 +53,7 @@ public class SslTransportTls12Tls13Test {
sslClientConfigs = clientCertStores.getTrustingConfig(serverCertStores);
LogContext logContext = new LogContext();
- ChannelBuilder channelBuilder = new SslChannelBuilder(Mode.CLIENT, null, false, logContext);
+ ChannelBuilder channelBuilder = new SslChannelBuilder(ConnectionMode.CLIENT, null, false, logContext);
channelBuilder.configure(sslClientConfigs);
this.selector = new Selector(5000, new Metrics(), TIME, "MetricGroup", channelBuilder, logContext);
}
@@ -159,7 +159,7 @@ public class SslTransportTls12Tls13Test {
}
private void createSelector(Map<String, Object> sslClientConfigs) {
- SslTransportLayerTest.TestSslChannelBuilder channelBuilder = new SslTransportLayerTest.TestSslChannelBuilder(Mode.CLIENT);
+ SslTransportLayerTest.TestSslChannelBuilder channelBuilder = new SslTransportLayerTest.TestSslChannelBuilder(ConnectionMode.CLIENT);
channelBuilder.configureBufferSizes(null, null, null);
channelBuilder.configure(sslClientConfigs);
if (this.selector != null) {
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SslVersionsTransportLayerTest.java b/clients/src/test/java/org/apache/kafka/common/network/SslVersionsTransportLayerTest.java
index 66ae5746386..be318bcc164 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SslVersionsTransportLayerTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SslVersionsTransportLayerTest.java
@@ -167,7 +167,7 @@ public class SslVersionsTransportLayerTest {
private Selector createClientSelector(Map<String, Object> sslClientConfigs) {
SslTransportLayerTest.TestSslChannelBuilder channelBuilder =
- new SslTransportLayerTest.TestSslChannelBuilder(Mode.CLIENT);
+ new SslTransportLayerTest.TestSslChannelBuilder(ConnectionMode.CLIENT);
channelBuilder.configureBufferSizes(null, null, null);
channelBuilder.configure(sslClientConfigs);
return new Selector(100 * 5000, new Metrics(), TIME, "MetricGroup", channelBuilder, new LogContext());
diff --git a/clients/src/test/java/org/apache/kafka/common/network/Tls12SelectorTest.java b/clients/src/test/java/org/apache/kafka/common/network/Tls12SelectorTest.java
index 82deb367b55..a0621cca072 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/Tls12SelectorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/Tls12SelectorTest.java
@@ -38,7 +38,7 @@ public class Tls12SelectorTest extends SslSelectorTest {
@Override
protected Map<String, Object> createSslClientConfigs(File trustStoreFile)
throws GeneralSecurityException, IOException {
- Map<String, Object> configs = TestSslUtils.createSslConfig(false, false, Mode.CLIENT,
+ Map<String, Object> configs = TestSslUtils.createSslConfig(false, false, ConnectionMode.CLIENT,
trustStoreFile, "client");
configs.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, Collections.singletonList("TLSv1.2"));
return configs;
diff --git a/clients/src/test/java/org/apache/kafka/common/network/Tls13SelectorTest.java b/clients/src/test/java/org/apache/kafka/common/network/Tls13SelectorTest.java
index ff521fade27..1ea9c43c01a 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/Tls13SelectorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/Tls13SelectorTest.java
@@ -43,7 +43,7 @@ public class Tls13SelectorTest extends SslSelectorTest {
@Override
protected Map<String, Object> createSslClientConfigs(File trustStoreFile) throws GeneralSecurityException, IOException {
- Map<String, Object> configs = TestSslUtils.createSslConfig(false, false, Mode.CLIENT,
+ Map<String, Object> configs = TestSslUtils.createSslConfig(false, false, ConnectionMode.CLIENT,
trustStoreFile, "client");
configs.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, Collections.singletonList("TLSv1.3"));
return configs;
diff --git a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
index 34afd41d175..883332f93e1 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslAuthenticatorTest.java
@@ -42,8 +42,8 @@ import org.apache.kafka.common.network.ChannelBuilder;
import org.apache.kafka.common.network.ChannelBuilders;
import org.apache.kafka.common.network.ChannelMetadataRegistry;
import org.apache.kafka.common.network.ChannelState;
+import org.apache.kafka.common.network.ConnectionMode;
import org.apache.kafka.common.network.ListenerName;
-import org.apache.kafka.common.network.Mode;
import org.apache.kafka.common.network.NetworkSend;
import org.apache.kafka.common.network.NetworkTestUtils;
import org.apache.kafka.common.network.NioEchoServer;
@@ -1672,7 +1672,7 @@ public class SaslAuthenticatorTest {
String saslMechanism = (String) saslClientConfigs.get(SaslConfigs.SASL_MECHANISM);
Map<String, ?> configs = new TestSecurityConfig(saslClientConfigs).values();
- this.channelBuilder = new AlternateSaslChannelBuilder(Mode.CLIENT,
+ this.channelBuilder = new AlternateSaslChannelBuilder(ConnectionMode.CLIENT,
Collections.singletonMap(saslMechanism, JaasContext.loadClientContext(configs)), securityProtocol, null,
false, saslMechanism, true, credentialCache, null, time);
this.channelBuilder.configure(configs);
@@ -1958,7 +1958,7 @@ public class SaslAuthenticatorTest {
return new ApiVersionsResponse(new ApiVersionsResponseData().setApiKeys(versionCollection));
};
- SaslChannelBuilder serverChannelBuilder = new SaslChannelBuilder(Mode.SERVER, jaasContexts,
+ SaslChannelBuilder serverChannelBuilder = new SaslChannelBuilder(ConnectionMode.SERVER, jaasContexts,
securityProtocol, listenerName, false, saslMechanism, true,
credentialCache, null, null, time, new LogContext(), apiVersionSupplier);
@@ -1999,7 +1999,7 @@ public class SaslAuthenticatorTest {
return new ApiVersionsResponse(data);
};
- SaslChannelBuilder serverChannelBuilder = new SaslChannelBuilder(Mode.SERVER, jaasContexts,
+ SaslChannelBuilder serverChannelBuilder = new SaslChannelBuilder(ConnectionMode.SERVER, jaasContexts,
securityProtocol, listenerName, false, saslMechanism, true,
credentialCache, null, null, time, new LogContext(), apiVersionSupplier) {
@Override
@@ -2034,7 +2034,7 @@ public class SaslAuthenticatorTest {
final JaasContext jaasContext = JaasContext.loadClientContext(configs);
final Map<String, JaasContext> jaasContexts = Collections.singletonMap(saslMechanism, jaasContext);
- SaslChannelBuilder clientChannelBuilder = new SaslChannelBuilder(Mode.CLIENT, jaasContexts,
+ SaslChannelBuilder clientChannelBuilder = new SaslChannelBuilder(ConnectionMode.CLIENT, jaasContexts,
securityProtocol, listenerName, false, saslMechanism, true,
null, null, null, time, new LogContext(), null) {
@@ -2571,11 +2571,11 @@ public class SaslAuthenticatorTest {
private static class AlternateSaslChannelBuilder extends SaslChannelBuilder {
private int numInvocations = 0;
- public AlternateSaslChannelBuilder(Mode mode, Map<String, JaasContext> jaasContexts,
+ public AlternateSaslChannelBuilder(ConnectionMode connectionMode, Map<String, JaasContext> jaasContexts,
SecurityProtocol securityProtocol, ListenerName listenerName, boolean isInterBrokerListener,
String clientSaslMechanism, boolean handshakeRequestEnable, CredentialCache credentialCache,
DelegationTokenCache tokenCache, Time time) {
- super(mode, jaasContexts, securityProtocol, listenerName, isInterBrokerListener, clientSaslMechanism,
+ super(connectionMode, jaasContexts, securityProtocol, listenerName, isInterBrokerListener, clientSaslMechanism,
handshakeRequestEnable, credentialCache, tokenCache, null, time, new LogContext(),
() -> TestUtils.defaultApiVersionsResponse(ApiMessageType.ListenerType.ZK_BROKER));
}
diff --git a/clients/src/test/java/org/apache/kafka/common/security/ssl/SslFactoryTest.java b/clients/src/test/java/org/apache/kafka/common/security/ssl/SslFactoryTest.java
index c185ba9d9fa..846425a5048 100644
--- a/clients/src/test/java/org/apache/kafka/common/security/ssl/SslFactoryTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/security/ssl/SslFactoryTest.java
@@ -21,7 +21,7 @@ import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.config.SecurityConfig;
import org.apache.kafka.common.config.SslConfigs;
import org.apache.kafka.common.config.types.Password;
-import org.apache.kafka.common.network.Mode;
+import org.apache.kafka.common.network.ConnectionMode;
import org.apache.kafka.common.security.TestSecurityConfig;
import org.apache.kafka.common.security.auth.SslEngineFactory;
import org.apache.kafka.common.security.ssl.DefaultSslEngineFactory.FileBasedStore;
@@ -73,10 +73,10 @@ public abstract class SslFactoryTest {
@Test
public void testSslFactoryConfiguration() throws Exception {
File trustStoreFile = TestUtils.tempFile("truststore", ".jks");
- Map<String, Object> serverSslConfig = sslConfigsBuilder(Mode.SERVER)
+ Map<String, Object> serverSslConfig = sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(trustStoreFile)
.build();
- try (SslFactory sslFactory = new SslFactory(Mode.SERVER, null, true)) {
+ try (SslFactory sslFactory = new SslFactory(ConnectionMode.SERVER, null, true)) {
sslFactory.configure(serverSslConfig);
//host and port are hints
SSLEngine engine = sslFactory.createSslEngine("localhost", 0);
@@ -91,7 +91,7 @@ public abstract class SslFactoryTest {
//generate server configs for keystore with multiple certificate chain
Map<String, Object> serverSslConfig = TestSslUtils.generateConfigsWithCertificateChains(tlsProtocol);
- try (SslFactory sslFactory = new SslFactory(Mode.SERVER, null, true)) {
+ try (SslFactory sslFactory = new SslFactory(ConnectionMode.SERVER, null, true)) {
sslFactory.configure(serverSslConfig);
SSLEngine engine = sslFactory.createSslEngine("localhost", 0);
assertNotNull(engine);
@@ -109,7 +109,7 @@ public abstract class SslFactoryTest {
tlsProtocol
);
serverSslConfig.put(SecurityConfig.SECURITY_PROVIDERS_CONFIG, testProviderCreator.getClass().getName());
- SslFactory sslFactory = new SslFactory(Mode.SERVER);
+ SslFactory sslFactory = new SslFactory(ConnectionMode.SERVER);
sslFactory.configure(serverSslConfig);
assertNotNull(sslFactory.sslEngineFactory(), "SslEngineFactory not created");
Security.removeProvider(testProviderCreator.getProvider().getName());
@@ -123,7 +123,7 @@ public abstract class SslFactoryTest {
TestTrustManagerFactory.ALGORITHM,
tlsProtocol
);
- SslFactory sslFactory = new SslFactory(Mode.SERVER);
+ SslFactory sslFactory = new SslFactory(ConnectionMode.SERVER);
assertThrows(KafkaException.class, () -> sslFactory.configure(serverSslConfig));
}
@@ -137,19 +137,19 @@ public abstract class SslFactoryTest {
);
serverSslConfig.put(SecurityConfig.SECURITY_PROVIDERS_CONFIG,
"com.fake.ProviderClass1,com.fake.ProviderClass2");
- SslFactory sslFactory = new SslFactory(Mode.SERVER);
+ SslFactory sslFactory = new SslFactory(ConnectionMode.SERVER);
assertThrows(KafkaException.class, () -> sslFactory.configure(serverSslConfig));
}
@Test
public void testSslFactoryWithoutPasswordConfiguration() throws Exception {
File trustStoreFile = TestUtils.tempFile("truststore", ".jks");
- Map<String, Object> serverSslConfig = sslConfigsBuilder(Mode.SERVER)
+ Map<String, Object> serverSslConfig = sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(trustStoreFile)
.build();
// unset the password
serverSslConfig.remove(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG);
- SslFactory sslFactory = new SslFactory(Mode.SERVER);
+ SslFactory sslFactory = new SslFactory(ConnectionMode.SERVER);
try {
sslFactory.configure(serverSslConfig);
} catch (Exception e) {
@@ -160,11 +160,11 @@ public abstract class SslFactoryTest {
@Test
public void testClientMode() throws Exception {
File trustStoreFile = TestUtils.tempFile("truststore", ".jks");
- Map<String, Object> clientSslConfig = sslConfigsBuilder(Mode.CLIENT)
+ Map<String, Object> clientSslConfig = sslConfigsBuilder(ConnectionMode.CLIENT)
.createNewTrustStore(trustStoreFile)
.useClientCert(false)
.build();
- SslFactory sslFactory = new SslFactory(Mode.CLIENT);
+ SslFactory sslFactory = new SslFactory(ConnectionMode.CLIENT);
sslFactory.configure(clientSslConfig);
//host and port are hints
SSLEngine engine = sslFactory.createSslEngine("localhost", 0);
@@ -174,19 +174,19 @@ public abstract class SslFactoryTest {
@Test
public void staleSslEngineFactoryShouldBeClosed() throws IOException, GeneralSecurityException {
File trustStoreFile = TestUtils.tempFile("truststore", ".jks");
- Map<String, Object> clientSslConfig = sslConfigsBuilder(Mode.SERVER)
+ Map<String, Object> clientSslConfig = sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(trustStoreFile)
.useClientCert(false)
.build();
clientSslConfig.put(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG, TestSslUtils.TestSslEngineFactory.class);
- SslFactory sslFactory = new SslFactory(Mode.SERVER);
+ SslFactory sslFactory = new SslFactory(ConnectionMode.SERVER);
sslFactory.configure(clientSslConfig);
TestSslUtils.TestSslEngineFactory sslEngineFactory = (TestSslUtils.TestSslEngineFactory) sslFactory.sslEngineFactory();
assertNotNull(sslEngineFactory);
assertFalse(sslEngineFactory.closed);
trustStoreFile = TestUtils.tempFile("truststore", ".jks");
- clientSslConfig = sslConfigsBuilder(Mode.SERVER)
+ clientSslConfig = sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(trustStoreFile)
.build();
clientSslConfig.put(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG, TestSslUtils.TestSslEngineFactory.class);
@@ -200,10 +200,10 @@ public abstract class SslFactoryTest {
@Test
public void testReconfiguration() throws Exception {
File trustStoreFile = TestUtils.tempFile("truststore", ".jks");
- Map<String, Object> sslConfig = sslConfigsBuilder(Mode.SERVER)
+ Map<String, Object> sslConfig = sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(trustStoreFile)
.build();
- SslFactory sslFactory = new SslFactory(Mode.SERVER);
+ SslFactory sslFactory = new SslFactory(ConnectionMode.SERVER);
sslFactory.configure(sslConfig);
SslEngineFactory sslEngineFactory = sslFactory.sslEngineFactory();
assertNotNull(sslEngineFactory, "SslEngineFactory not created");
@@ -215,7 +215,7 @@ public abstract class SslFactoryTest {
// Verify that the SslEngineFactory is recreated on reconfigure() if config is changed
trustStoreFile = TestUtils.tempFile("truststore", ".jks");
- sslConfig = sslConfigsBuilder(Mode.SERVER)
+ sslConfig = sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(trustStoreFile)
.build();
sslFactory.reconfigure(sslConfig);
@@ -252,13 +252,13 @@ public abstract class SslFactoryTest {
@Test
public void testReconfigurationWithoutTruststore() throws Exception {
File trustStoreFile = TestUtils.tempFile("truststore", ".jks");
- Map<String, Object> sslConfig = sslConfigsBuilder(Mode.SERVER)
+ Map<String, Object> sslConfig = sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(trustStoreFile)
.build();
sslConfig.remove(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG);
sslConfig.remove(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG);
sslConfig.remove(SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG);
- SslFactory sslFactory = new SslFactory(Mode.SERVER);
+ SslFactory sslFactory = new SslFactory(ConnectionMode.SERVER);
sslFactory.configure(sslConfig);
SSLContext sslContext = ((DefaultSslEngineFactory) sslFactory.sslEngineFactory()).sslContext();
assertNotNull(sslContext, "SSL context not created");
@@ -266,7 +266,7 @@ public abstract class SslFactoryTest {
"SSL context recreated unnecessarily");
assertFalse(sslFactory.createSslEngine("localhost", 0).getUseClientMode());
- Map<String, Object> sslConfig2 = sslConfigsBuilder(Mode.SERVER)
+ Map<String, Object> sslConfig2 = sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(trustStoreFile)
.build();
try {
@@ -280,13 +280,13 @@ public abstract class SslFactoryTest {
@Test
public void testReconfigurationWithoutKeystore() throws Exception {
File trustStoreFile = TestUtils.tempFile("truststore", ".jks");
- Map<String, Object> sslConfig = sslConfigsBuilder(Mode.SERVER)
+ Map<String, Object> sslConfig = sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(trustStoreFile)
.build();
sslConfig.remove(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG);
sslConfig.remove(SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG);
sslConfig.remove(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG);
- SslFactory sslFactory = new SslFactory(Mode.SERVER);
+ SslFactory sslFactory = new SslFactory(ConnectionMode.SERVER);
sslFactory.configure(sslConfig);
SSLContext sslContext = ((DefaultSslEngineFactory) sslFactory.sslEngineFactory()).sslContext();
assertNotNull(sslContext, "SSL context not created");
@@ -295,7 +295,7 @@ public abstract class SslFactoryTest {
assertFalse(sslFactory.createSslEngine("localhost", 0).getUseClientMode());
File newTrustStoreFile = TestUtils.tempFile("truststore", ".jks");
- sslConfig = sslConfigsBuilder(Mode.SERVER)
+ sslConfig = sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(newTrustStoreFile)
.build();
sslConfig.remove(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG);
@@ -305,7 +305,7 @@ public abstract class SslFactoryTest {
assertNotSame(sslContext, ((DefaultSslEngineFactory) sslFactory.sslEngineFactory()).sslContext(),
"SSL context not recreated");
- sslConfig = sslConfigsBuilder(Mode.SERVER)
+ sslConfig = sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(newTrustStoreFile)
.build();
try {
@@ -319,13 +319,13 @@ public abstract class SslFactoryTest {
@Test
public void testPemReconfiguration() throws Exception {
Properties props = new Properties();
- props.putAll(sslConfigsBuilder(Mode.SERVER)
+ props.putAll(sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(null)
.usePem(true)
.build());
TestSecurityConfig sslConfig = new TestSecurityConfig(props);
- SslFactory sslFactory = new SslFactory(Mode.SERVER);
+ SslFactory sslFactory = new SslFactory(ConnectionMode.SERVER);
sslFactory.configure(sslConfig.values());
SslEngineFactory sslEngineFactory = sslFactory.sslEngineFactory();
assertNotNull(sslEngineFactory, "SslEngineFactory not created");
@@ -360,10 +360,10 @@ public abstract class SslFactoryTest {
@Test
public void testKeyStoreTrustStoreValidation() throws Exception {
File trustStoreFile = TestUtils.tempFile("truststore", ".jks");
- Map<String, Object> serverSslConfig = sslConfigsBuilder(Mode.SERVER)
+ Map<String, Object> serverSslConfig = sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(trustStoreFile)
.build();
- SslFactory sslFactory = new SslFactory(Mode.SERVER);
+ SslFactory sslFactory = new SslFactory(ConnectionMode.SERVER);
sslFactory.configure(serverSslConfig);
assertNotNull(sslFactory.sslEngineFactory(), "SslEngineFactory not created");
}
@@ -372,13 +372,13 @@ public abstract class SslFactoryTest {
public void testUntrustedKeyStoreValidationFails() throws Exception {
File trustStoreFile1 = TestUtils.tempFile("truststore1", ".jks");
File trustStoreFile2 = TestUtils.tempFile("truststore2", ".jks");
- Map<String, Object> sslConfig1 = sslConfigsBuilder(Mode.SERVER)
+ Map<String, Object> sslConfig1 = sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(trustStoreFile1)
.build();
- Map<String, Object> sslConfig2 = sslConfigsBuilder(Mode.SERVER)
+ Map<String, Object> sslConfig2 = sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(trustStoreFile2)
.build();
- SslFactory sslFactory = new SslFactory(Mode.SERVER, null, true);
+ SslFactory sslFactory = new SslFactory(ConnectionMode.SERVER, null, true);
for (String key : Arrays.asList(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG,
SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG,
SslConfigs.SSL_TRUSTSTORE_TYPE_CONFIG,
@@ -405,15 +405,15 @@ public abstract class SslFactoryTest {
private void verifyKeystoreVerifiableUsingTruststore(boolean usePem, String tlsProtocol) throws Exception {
File trustStoreFile1 = usePem ? null : TestUtils.tempFile("truststore1", ".jks");
- Map<String, Object> sslConfig1 = sslConfigsBuilder(Mode.SERVER)
+ Map<String, Object> sslConfig1 = sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(trustStoreFile1)
.usePem(usePem)
.build();
- SslFactory sslFactory = new SslFactory(Mode.SERVER, null, true);
+ SslFactory sslFactory = new SslFactory(ConnectionMode.SERVER, null, true);
sslFactory.configure(sslConfig1);
File trustStoreFile2 = usePem ? null : TestUtils.tempFile("truststore2", ".jks");
- Map<String, Object> sslConfig2 = sslConfigsBuilder(Mode.SERVER)
+ Map<String, Object> sslConfig2 = sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(trustStoreFile2)
.usePem(usePem)
.build();
@@ -441,12 +441,12 @@ public abstract class SslFactoryTest {
private void verifyCertificateEntriesValidation(boolean usePem, String tlsProtocol) throws Exception {
File trustStoreFile = usePem ? null : TestUtils.tempFile("truststore", ".jks");
- Map<String, Object> serverSslConfig = sslConfigsBuilder(Mode.SERVER)
+ Map<String, Object> serverSslConfig = sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(trustStoreFile)
.usePem(usePem)
.build();
File newTrustStoreFile = usePem ? null : TestUtils.tempFile("truststore", ".jks");
- Map<String, Object> newCnConfig = sslConfigsBuilder(Mode.SERVER)
+ Map<String, Object> newCnConfig = sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(newTrustStoreFile)
.cn("Another CN")
.usePem(usePem)
@@ -469,12 +469,12 @@ public abstract class SslFactoryTest {
@Test
public void testClientSpecifiedSslEngineFactoryUsed() throws Exception {
File trustStoreFile = TestUtils.tempFile("truststore", ".jks");
- Map<String, Object> clientSslConfig = sslConfigsBuilder(Mode.CLIENT)
+ Map<String, Object> clientSslConfig = sslConfigsBuilder(ConnectionMode.CLIENT)
.createNewTrustStore(trustStoreFile)
.useClientCert(false)
.build();
clientSslConfig.put(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG, TestSslUtils.TestSslEngineFactory.class);
- SslFactory sslFactory = new SslFactory(Mode.CLIENT);
+ SslFactory sslFactory = new SslFactory(ConnectionMode.CLIENT);
sslFactory.configure(clientSslConfig);
assertInstanceOf(TestSslUtils.TestSslEngineFactory.class, sslFactory.sslEngineFactory(),
"SslEngineFactory must be of expected type");
@@ -483,12 +483,12 @@ public abstract class SslFactoryTest {
@Test
public void testEngineFactoryClosed() throws Exception {
File trustStoreFile = TestUtils.tempFile("truststore", ".jks");
- Map<String, Object> clientSslConfig = sslConfigsBuilder(Mode.CLIENT)
+ Map<String, Object> clientSslConfig = sslConfigsBuilder(ConnectionMode.CLIENT)
.createNewTrustStore(trustStoreFile)
.useClientCert(false)
.build();
clientSslConfig.put(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG, TestSslUtils.TestSslEngineFactory.class);
- SslFactory sslFactory = new SslFactory(Mode.CLIENT);
+ SslFactory sslFactory = new SslFactory(ConnectionMode.CLIENT);
sslFactory.configure(clientSslConfig);
TestSslUtils.TestSslEngineFactory engine = (TestSslUtils.TestSslEngineFactory) sslFactory.sslEngineFactory();
assertFalse(engine.closed);
@@ -502,12 +502,12 @@ public abstract class SslFactoryTest {
@Test
public void testServerSpecifiedSslEngineFactoryUsed() throws Exception {
File trustStoreFile = TestUtils.tempFile("truststore", ".jks");
- Map<String, Object> serverSslConfig = sslConfigsBuilder(Mode.SERVER)
+ Map<String, Object> serverSslConfig = sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(trustStoreFile)
.useClientCert(false)
.build();
serverSslConfig.put(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG, TestSslUtils.TestSslEngineFactory.class);
- SslFactory sslFactory = new SslFactory(Mode.SERVER);
+ SslFactory sslFactory = new SslFactory(ConnectionMode.SERVER);
sslFactory.configure(serverSslConfig);
assertInstanceOf(TestSslUtils.TestSslEngineFactory.class, sslFactory.sslEngineFactory(),
"SslEngineFactory must be of expected type");
@@ -519,24 +519,24 @@ public abstract class SslFactoryTest {
@Test
public void testInvalidSslEngineFactory() throws Exception {
File trustStoreFile = TestUtils.tempFile("truststore", ".jks");
- Map<String, Object> clientSslConfig = sslConfigsBuilder(Mode.CLIENT)
+ Map<String, Object> clientSslConfig = sslConfigsBuilder(ConnectionMode.CLIENT)
.createNewTrustStore(trustStoreFile)
.useClientCert(false)
.build();
clientSslConfig.put(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG, String.class);
- SslFactory sslFactory = new SslFactory(Mode.CLIENT);
+ SslFactory sslFactory = new SslFactory(ConnectionMode.CLIENT);
assertThrows(ClassCastException.class, () -> sslFactory.configure(clientSslConfig));
}
@Test
public void testUsedConfigs() throws IOException, GeneralSecurityException {
- Map<String, Object> serverSslConfig = sslConfigsBuilder(Mode.SERVER)
+ Map<String, Object> serverSslConfig = sslConfigsBuilder(ConnectionMode.SERVER)
.createNewTrustStore(TestUtils.tempFile("truststore", ".jks"))
.useClientCert(false)
.build();
serverSslConfig.put(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG, TestSslUtils.TestSslEngineFactory.class);
TestSecurityConfig securityConfig = new TestSecurityConfig(serverSslConfig);
- SslFactory sslFactory = new SslFactory(Mode.SERVER);
+ SslFactory sslFactory = new SslFactory(ConnectionMode.SERVER);
sslFactory.configure(securityConfig.values());
assertFalse(securityConfig.unused().contains(SslConfigs.SSL_ENGINE_FACTORY_CLASS_CONFIG));
}
@@ -603,7 +603,7 @@ public abstract class SslFactoryTest {
return store.get();
}
- private TestSslUtils.SslConfigsBuilder sslConfigsBuilder(Mode mode) {
- return new TestSslUtils.SslConfigsBuilder(mode).tlsProtocol(tlsProtocol);
+ private TestSslUtils.SslConfigsBuilder sslConfigsBuilder(ConnectionMode connectionMode) {
+ return new TestSslUtils.SslConfigsBuilder(connectionMode).tlsProtocol(tlsProtocol);
}
}
diff --git a/clients/src/test/java/org/apache/kafka/test/TestSslUtils.java b/clients/src/test/java/org/apache/kafka/test/TestSslUtils.java
index d76a409575d..72a1ccfe651 100644
--- a/clients/src/test/java/org/apache/kafka/test/TestSslUtils.java
+++ b/clients/src/test/java/org/apache/kafka/test/TestSslUtils.java
@@ -18,7 +18,7 @@ package org.apache.kafka.test;
import org.apache.kafka.common.config.SslConfigs;
import org.apache.kafka.common.config.types.Password;
-import org.apache.kafka.common.network.Mode;
+import org.apache.kafka.common.network.ConnectionMode;
import org.apache.kafka.common.security.auth.SslEngineFactory;
import org.apache.kafka.common.security.ssl.DefaultSslEngineFactory;
@@ -219,21 +219,21 @@ public class TestSslUtils {
return sslConfigs;
}
- public static Map<String, Object> createSslConfig(boolean useClientCert, boolean trustStore, Mode mode, File trustStoreFile, String certAlias)
+ public static Map<String, Object> createSslConfig(boolean useClientCert, boolean trustStore, ConnectionMode connectionMode, File trustStoreFile, String certAlias)
throws IOException, GeneralSecurityException {
- return createSslConfig(useClientCert, trustStore, mode, trustStoreFile, certAlias, "localhost");
+ return createSslConfig(useClientCert, trustStore, connectionMode, trustStoreFile, certAlias, "localhost");
}
public static Map<String, Object> createSslConfig(boolean useClientCert, boolean trustStore,
- Mode mode, File trustStoreFile, String certAlias, String cn)
+ ConnectionMode connectionMode, File trustStoreFile, String certAlias, String cn)
throws IOException, GeneralSecurityException {
- return createSslConfig(useClientCert, trustStore, mode, trustStoreFile, certAlias, cn, new CertificateBuilder());
+ return createSslConfig(useClientCert, trustStore, connectionMode, trustStoreFile, certAlias, cn, new CertificateBuilder());
}
public static Map<String, Object> createSslConfig(boolean useClientCert, boolean createTrustStore,
- Mode mode, File trustStoreFile, String certAlias, String cn, CertificateBuilder certBuilder)
+ ConnectionMode connectionMode, File trustStoreFile, String certAlias, String cn, CertificateBuilder certBuilder)
throws IOException, GeneralSecurityException {
- SslConfigsBuilder builder = new SslConfigsBuilder(mode)
+ SslConfigsBuilder builder = new SslConfigsBuilder(connectionMode)
.useClientCert(useClientCert)
.certAlias(certAlias)
.cn(cn)
@@ -565,7 +565,7 @@ public class TestSslUtils {
}
public static class SslConfigsBuilder {
- final Mode mode;
+ final ConnectionMode connectionMode;
String tlsProtocol;
boolean useClientCert;
boolean createTrustStore;
@@ -579,15 +579,15 @@ public class TestSslUtils {
CertificateBuilder certBuilder;
boolean usePem;
- public SslConfigsBuilder(Mode mode) {
- this.mode = mode;
+ public SslConfigsBuilder(ConnectionMode connectionMode) {
+ this.connectionMode = connectionMode;
this.tlsProtocol = DEFAULT_TLS_PROTOCOL_FOR_TESTS;
trustStorePassword = new Password(TRUST_STORE_PASSWORD);
- keyStorePassword = mode == Mode.SERVER ? new Password("ServerPassword") : new Password("ClientPassword");
+ keyStorePassword = connectionMode == ConnectionMode.SERVER ? new Password("ServerPassword") : new Password("ClientPassword");
keyPassword = keyStorePassword;
this.certBuilder = new CertificateBuilder();
this.cn = "localhost";
- this.certAlias = mode.name().toLowerCase(Locale.ROOT);
+ this.certAlias = connectionMode.name().toLowerCase(Locale.ROOT);
this.algorithm = "RSA";
this.createTrustStore = true;
}
@@ -650,13 +650,13 @@ public class TestSslUtils {
Map<String, X509Certificate> certs = new HashMap<>();
File keyStoreFile = null;
- if (mode == Mode.CLIENT && useClientCert) {
+ if (connectionMode == ConnectionMode.CLIENT && useClientCert) {
keyStoreFile = TestUtils.tempFile("clientKS", ".jks");
KeyPair cKP = generateKeyPair(algorithm);
X509Certificate cCert = certBuilder.generate("CN=" + cn + ", O=A client", cKP);
createKeyStore(keyStoreFile.getPath(), keyStorePassword, keyPassword, "client", cKP.getPrivate(), cCert);
certs.put(certAlias, cCert);
- } else if (mode == Mode.SERVER) {
+ } else if (connectionMode == ConnectionMode.SERVER) {
keyStoreFile = TestUtils.tempFile("serverKS", ".jks");
KeyPair sKP = generateKeyPair(algorithm);
X509Certificate sCert = certBuilder.generate("CN=" + cn + ", O=A server", sKP);
@@ -674,7 +674,7 @@ public class TestSslUtils {
sslConfigs.put(SslConfigs.SSL_PROTOCOL_CONFIG, tlsProtocol); // protocol to create SSLContext
- if (mode == Mode.SERVER || (mode == Mode.CLIENT && keyStoreFile != null)) {
+ if (connectionMode == ConnectionMode.SERVER || (connectionMode == ConnectionMode.CLIENT && keyStoreFile != null)) {
sslConfigs.put(SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, keyStoreFile.getPath());
sslConfigs.put(SslConfigs.SSL_KEYSTORE_TYPE_CONFIG, "JKS");
sslConfigs.put(SslConfigs.SSL_KEYMANAGER_ALGORITHM_CONFIG, TrustManagerFactory.getDefaultAlgorithm());
@@ -703,9 +703,9 @@ public class TestSslUtils {
sslConfigs.put(SslConfigs.SSL_PROTOCOL_CONFIG, tlsProtocol);
sslConfigs.put(SslConfigs.SSL_ENABLED_PROTOCOLS_CONFIG, Collections.singletonList(tlsProtocol));
- if (mode != Mode.CLIENT || useClientCert) {
+ if (connectionMode != ConnectionMode.CLIENT || useClientCert) {
KeyPair keyPair = generateKeyPair(algorithm);
- X509Certificate cert = certBuilder.generate("CN=" + cn + ", O=A " + mode.name().toLowerCase(Locale.ROOT), keyPair);
+ X509Certificate cert = certBuilder.generate("CN=" + cn + ", O=A " + connectionMode.name().toLowerCase(Locale.ROOT), keyPair);
Password privateKeyPem = new Password(pem(keyPair.getPrivate(), keyPassword));
Password certPem = new Password(pem(cert));
diff --git a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationSSLTest.java b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationSSLTest.java
index 18f734a00a2..7999f5f9a4a 100644
--- a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationSSLTest.java
+++ b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationSSLTest.java
@@ -19,7 +19,7 @@ package org.apache.kafka.connect.mirror.integration;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.config.SslConfigs;
import org.apache.kafka.common.config.types.Password;
-import org.apache.kafka.common.network.Mode;
+import org.apache.kafka.common.network.ConnectionMode;
import org.apache.kafka.network.SocketServerConfigs;
import org.apache.kafka.server.config.ReplicationConfigs;
import org.apache.kafka.test.TestSslUtils;
@@ -40,7 +40,7 @@ public class MirrorConnectorsIntegrationSSLTest extends MirrorConnectorsIntegrat
@BeforeEach
public void startClusters() throws Exception {
- Map<String, Object> sslConfig = TestSslUtils.createSslConfig(false, true, Mode.SERVER, TestUtils.tempFile(), "testCert");
+ Map<String, Object> sslConfig = TestSslUtils.createSslConfig(false, true, ConnectionMode.SERVER, TestUtils.tempFile(), "testCert");
// enable SSL on backup kafka broker
backupBrokerProps.put(SocketServerConfigs.LISTENERS_CONFIG, "SSL://localhost:0");
backupBrokerProps.put(ReplicationConfigs.INTER_BROKER_LISTENER_NAME_CONFIG, "SSL");
diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/RestForwardingIntegrationTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/RestForwardingIntegrationTest.java
index c471cfa823a..81b18d03442 100644
--- a/connect/runtime/src/test/java/org/apache/kafka/connect/integration/RestForwardingIntegrationTest.java
+++ b/connect/runtime/src/test/java/org/apache/kafka/connect/integration/RestForwardingIntegrationTest.java
@@ -18,7 +18,7 @@ package org.apache.kafka.connect.integration;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.config.types.Password;
-import org.apache.kafka.common.network.Mode;
+import org.apache.kafka.common.network.ConnectionMode;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.connect.runtime.Herder;
import org.apache.kafka.connect.runtime.WorkerConfig;
@@ -99,7 +99,7 @@ public class RestForwardingIntegrationTest {
@BeforeEach
public void setUp() throws IOException, GeneralSecurityException {
- sslConfig = TestSslUtils.createSslConfig(false, true, Mode.SERVER, TestUtils.tempFile(), "testCert");
+ sslConfig = TestSslUtils.createSslConfig(false, true, ConnectionMode.SERVER, TestUtils.tempFile(), "testCert");
responses = new ArrayList<>();
}
diff --git a/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala b/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala
index 6804dc4c38b..6e8341d7358 100644
--- a/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala
+++ b/core/src/test/scala/integration/kafka/api/BaseProducerSendTest.scala
@@ -29,7 +29,7 @@ import org.apache.kafka.clients.consumer.Consumer
import org.apache.kafka.clients.producer._
import org.apache.kafka.common.config.TopicConfig
import org.apache.kafka.common.errors.TimeoutException
-import org.apache.kafka.common.network.{ListenerName, Mode}
+import org.apache.kafka.common.network.{ListenerName, ConnectionMode}
import org.apache.kafka.common.record.TimestampType
import org.apache.kafka.common.security.auth.SecurityProtocol
import org.apache.kafka.common.{KafkaException, TopicPartition}
@@ -70,7 +70,7 @@ abstract class BaseProducerSendTest extends KafkaServerTestHarness {
super.setUp(testInfo)
admin = TestUtils.createAdminClient(brokers, listenerName,
- TestUtils.securityConfigs(Mode.CLIENT,
+ TestUtils.securityConfigs(ConnectionMode.CLIENT,
securityProtocol,
trustStoreFile,
"adminClient",
diff --git a/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala b/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala
index 192dfa392d9..bbdc01e72aa 100644
--- a/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala
+++ b/core/src/test/scala/integration/kafka/api/IntegrationTestHarness.scala
@@ -27,7 +27,7 @@ import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig}
import kafka.server.KafkaConfig
import kafka.integration.KafkaServerTestHarness
import org.apache.kafka.clients.admin.{Admin, AdminClientConfig}
-import org.apache.kafka.common.network.{ListenerName, Mode}
+import org.apache.kafka.common.network.{ListenerName, ConnectionMode}
import org.apache.kafka.common.serialization.{ByteArrayDeserializer, ByteArraySerializer, Deserializer, Serializer}
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig
import org.apache.kafka.network.SocketServerConfigs
@@ -162,7 +162,7 @@ abstract class IntegrationTestHarness extends KafkaServerTestHarness {
}
def clientSecurityProps(certAlias: String): Properties = {
- TestUtils.securityConfigs(Mode.CLIENT, securityProtocol, trustStoreFile, certAlias, TestUtils.SslCertificateCn,
+ TestUtils.securityConfigs(ConnectionMode.CLIENT, securityProtocol, trustStoreFile, certAlias, TestUtils.SslCertificateCn,
clientSaslProperties)
}
diff --git a/core/src/test/scala/integration/kafka/api/SaslPlainSslEndToEndAuthorizationTest.scala b/core/src/test/scala/integration/kafka/api/SaslPlainSslEndToEndAuthorizationTest.scala
index 0ae3c437cf6..b4772aa3cc5 100644
--- a/core/src/test/scala/integration/kafka/api/SaslPlainSslEndToEndAuthorizationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/SaslPlainSslEndToEndAuthorizationTest.scala
@@ -22,7 +22,7 @@ import kafka.utils.TestUtils.isAclSecure
import kafka.zk.ZkData
import org.apache.kafka.common.config.SaslConfigs
import org.apache.kafka.common.config.internals.BrokerSecurityConfigs
-import org.apache.kafka.common.network.Mode
+import org.apache.kafka.common.network.ConnectionMode
import org.apache.kafka.common.security.auth._
import org.apache.kafka.common.security.authenticator.DefaultKafkaPrincipalBuilder
import org.apache.kafka.common.security.plain.PlainAuthenticateCallback
@@ -139,7 +139,7 @@ class SaslPlainSslEndToEndAuthorizationTest extends SaslEndToEndAuthorizationTes
// Generate SSL certificates for clients since we are enabling TLS mutual authentication
// in this test for the SASL_SSL listener.
override def clientSecurityProps(certAlias: String): Properties = {
- TestUtils.securityConfigs(Mode.CLIENT, securityProtocol, trustStoreFile, certAlias, TestUtils.SslCertificateCn,
+ TestUtils.securityConfigs(ConnectionMode.CLIENT, securityProtocol, trustStoreFile, certAlias, TestUtils.SslCertificateCn,
clientSaslProperties, needsClientCert = Some(true))
}
diff --git a/core/src/test/scala/integration/kafka/api/SslAdminIntegrationTest.scala b/core/src/test/scala/integration/kafka/api/SslAdminIntegrationTest.scala
index 58c131556d7..fdaafce2659 100644
--- a/core/src/test/scala/integration/kafka/api/SslAdminIntegrationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/SslAdminIntegrationTest.scala
@@ -28,7 +28,7 @@ import org.apache.kafka.common.resource.{PatternType, ResourcePattern, ResourceT
import org.apache.kafka.common.security.auth.{AuthenticationContext, KafkaPrincipal, SecurityProtocol, SslAuthenticationContext}
import org.apache.kafka.common.security.authenticator.DefaultKafkaPrincipalBuilder
import org.apache.kafka.server.authorizer._
-import org.apache.kafka.common.network.Mode
+import org.apache.kafka.common.network.ConnectionMode
import org.apache.kafka.server.metrics.KafkaYammerMetrics
import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse, assertNotNull, assertTrue}
import org.junit.jupiter.api.{AfterEach, Test}
@@ -284,7 +284,7 @@ class SslAdminIntegrationTest extends SaslSslAdminIntegrationTest {
// Override the CN to create a principal based on it
override def superuserSecurityProps(certAlias: String): Properties = {
- val props = TestUtils.securityConfigs(Mode.CLIENT, securityProtocol, trustStoreFile, certAlias, SslAdminIntegrationTest.superuserCn,
+ val props = TestUtils.securityConfigs(ConnectionMode.CLIENT, securityProtocol, trustStoreFile, certAlias, SslAdminIntegrationTest.superuserCn,
clientSaslProperties)
props.remove(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG)
props
@@ -292,7 +292,7 @@ class SslAdminIntegrationTest extends SaslSslAdminIntegrationTest {
// Override the CN to create a principal based on it
override def clientSecurityProps(certAlias: String): Properties = {
- val props = TestUtils.securityConfigs(Mode.CLIENT, securityProtocol, trustStoreFile, certAlias, SslAdminIntegrationTest.clientCn,
+ val props = TestUtils.securityConfigs(ConnectionMode.CLIENT, securityProtocol, trustStoreFile, certAlias, SslAdminIntegrationTest.clientCn,
clientSaslProperties)
props.remove(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG)
props
diff --git a/core/src/test/scala/integration/kafka/api/SslEndToEndAuthorizationTest.scala b/core/src/test/scala/integration/kafka/api/SslEndToEndAuthorizationTest.scala
index 272b0cfb3c0..5c5c24d36cc 100644
--- a/core/src/test/scala/integration/kafka/api/SslEndToEndAuthorizationTest.scala
+++ b/core/src/test/scala/integration/kafka/api/SslEndToEndAuthorizationTest.scala
@@ -22,7 +22,7 @@ import java.util.Properties
import kafka.utils.TestUtils
import org.apache.kafka.common.config.SslConfigs
import org.apache.kafka.common.config.internals.BrokerSecurityConfigs
-import org.apache.kafka.common.network.Mode
+import org.apache.kafka.common.network.ConnectionMode
import org.apache.kafka.common.security.auth._
import org.apache.kafka.common.security.authenticator.DefaultKafkaPrincipalBuilder
import org.apache.kafka.common.utils.Java
@@ -78,7 +78,7 @@ class SslEndToEndAuthorizationTest extends EndToEndAuthorizationTest {
}
override def clientSecurityProps(certAlias: String): Properties = {
- val props = TestUtils.securityConfigs(Mode.CLIENT, securityProtocol, trustStoreFile,
+ val props = TestUtils.securityConfigs(ConnectionMode.CLIENT, securityProtocol, trustStoreFile,
certAlias, clientCn, clientSaslProperties, tlsProtocol)
props.remove(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG)
props
@@ -86,7 +86,7 @@ class SslEndToEndAuthorizationTest extends EndToEndAuthorizationTest {
// This test doesn't really care about matching the SSL certificate to a particular principal
// We can override the CN and create a principal based on it or on the server SSL
override def superuserSecurityProps(certAlias: String): Properties = {
- val props = TestUtils.securityConfigs(Mode.CLIENT, securityProtocol, trustStoreFile,
+ val props = TestUtils.securityConfigs(ConnectionMode.CLIENT, securityProtocol, trustStoreFile,
certAlias, superuserCn, clientSaslProperties, tlsProtocol)
props.remove(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG)
props
diff --git a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
index a9d7fb1e3fb..32ad28322cc 100644
--- a/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
+++ b/core/src/test/scala/integration/kafka/server/DynamicBrokerReconfigurationTest.scala
@@ -53,7 +53,7 @@ import org.apache.kafka.common.errors.{AuthenticationException, InvalidRequestEx
import org.apache.kafka.common.internals.Topic
import org.apache.kafka.common.message.MetadataRequestData
import org.apache.kafka.common.metrics.{KafkaMetric, MetricsContext, MetricsReporter, Quota}
-import org.apache.kafka.common.network.{ListenerName, Mode}
+import org.apache.kafka.common.network.{ListenerName, ConnectionMode}
import org.apache.kafka.common.network.CertStores.{KEYSTORE_PROPS, TRUSTSTORE_PROPS}
import org.apache.kafka.common.record.TimestampType
import org.apache.kafka.common.requests.MetadataRequest
@@ -106,8 +106,8 @@ class DynamicBrokerReconfigurationTest extends QuorumTestHarness with SaslSetup
private val trustStoreFile1 = TestUtils.tempFile("truststore", ".jks")
private val trustStoreFile2 = TestUtils.tempFile("truststore", ".jks")
- private val sslProperties1 = TestUtils.sslConfigs(Mode.SERVER, clientCert = false, Some(trustStoreFile1), "kafka")
- private val sslProperties2 = TestUtils.sslConfigs(Mode.SERVER, clientCert = false, Some(trustStoreFile2), "kafka")
+ private val sslProperties1 = TestUtils.sslConfigs(ConnectionMode.SERVER, clientCert = false, Some(trustStoreFile1), "kafka")
+ private val sslProperties2 = TestUtils.sslConfigs(ConnectionMode.SERVER, clientCert = false, Some(trustStoreFile2), "kafka")
private val invalidSslProperties = invalidSslConfigs
@BeforeEach
diff --git a/core/src/test/scala/integration/kafka/server/GssapiAuthenticationTest.scala b/core/src/test/scala/integration/kafka/server/GssapiAuthenticationTest.scala
index c507d46a8eb..f2b7b0cbe08 100644
--- a/core/src/test/scala/integration/kafka/server/GssapiAuthenticationTest.scala
+++ b/core/src/test/scala/integration/kafka/server/GssapiAuthenticationTest.scala
@@ -262,7 +262,7 @@ class GssapiAuthenticationTest extends IntegrationTestHarness with SaslSetup {
clientConfig.setProperty(SaslConfigs.SASL_KERBEROS_MIN_TIME_BEFORE_RELOGIN, "0")
val config = new TestSecurityConfig(clientConfig)
val jaasContexts = Collections.singletonMap("GSSAPI", JaasContext.loadClientContext(config.values()))
- val channelBuilder = new SaslChannelBuilder(Mode.CLIENT, jaasContexts, securityProtocol,
+ val channelBuilder = new SaslChannelBuilder(ConnectionMode.CLIENT, jaasContexts, securityProtocol,
null, false, kafkaClientSaslMechanism, true, null, null, null, time, new LogContext(),
() => org.apache.kafka.test.TestUtils.defaultApiVersionsResponse(ListenerType.ZK_BROKER)) {
override protected def defaultLoginClass(): Class[_ <: Login] = classOf[TestableKerberosLogin]
diff --git a/core/src/test/scala/integration/kafka/server/IntegrationTestUtils.scala b/core/src/test/scala/integration/kafka/server/IntegrationTestUtils.scala
index 6ec24a7bf41..0623a16b388 100644
--- a/core/src/test/scala/integration/kafka/server/IntegrationTestUtils.scala
+++ b/core/src/test/scala/integration/kafka/server/IntegrationTestUtils.scala
@@ -26,7 +26,7 @@ import kafka.network.SocketServer
import kafka.utils.Implicits._
import kafka.utils.TestUtils
import org.apache.kafka.clients.admin.{Admin, NewTopic}
-import org.apache.kafka.common.network.{ListenerName, Mode}
+import org.apache.kafka.common.network.{ListenerName, ConnectionMode}
import org.apache.kafka.common.protocol.ApiKeys
import org.apache.kafka.common.requests.{AbstractRequest, AbstractResponse, RequestHeader, ResponseHeader}
import org.apache.kafka.common.security.auth.SecurityProtocol
@@ -139,6 +139,6 @@ object IntegrationTestUtils {
}
def clientSecurityProps(certAlias: String): Properties = {
- TestUtils.securityConfigs(Mode.CLIENT, securityProtocol, None, certAlias, TestUtils.SslCertificateCn, None) // TODO use real trust store and client SASL properties
+ TestUtils.securityConfigs(ConnectionMode.CLIENT, securityProtocol, None, certAlias, TestUtils.SslCertificateCn, None) // TODO use real trust store and client SASL properties
}
}
diff --git a/core/src/test/scala/integration/kafka/server/MultipleListenersWithSameSecurityProtocolBaseTest.scala b/core/src/test/scala/integration/kafka/server/MultipleListenersWithSameSecurityProtocolBaseTest.scala
index fed241f5d7b..ec29d3d4d94 100644
--- a/core/src/test/scala/integration/kafka/server/MultipleListenersWithSameSecurityProtocolBaseTest.scala
+++ b/core/src/test/scala/integration/kafka/server/MultipleListenersWithSameSecurityProtocolBaseTest.scala
@@ -29,7 +29,7 @@ import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.config.internals.BrokerSecurityConfigs
import org.apache.kafka.common.config.{SaslConfigs, SslConfigs}
import org.apache.kafka.common.internals.Topic
-import org.apache.kafka.common.network.{ListenerName, Mode}
+import org.apache.kafka.common.network.{ListenerName, ConnectionMode}
import org.apache.kafka.server.config.{ReplicationConfigs, ZkConfigs}
import org.apache.kafka.coordinator.group.GroupCoordinatorConfig
import org.apache.kafka.network.SocketServerConfigs
@@ -91,7 +91,7 @@ abstract class MultipleListenersWithSameSecurityProtocolBaseTest extends QuorumT
props.put(SaslConfigs.SASL_KERBEROS_SERVICE_NAME, "kafka")
props ++= dynamicJaasSections
- props ++= TestUtils.sslConfigs(Mode.SERVER, clientCert = false, Some(trustStoreFile), s"server$brokerId")
+ props ++= TestUtils.sslConfigs(ConnectionMode.SERVER, clientCert = false, Some(trustStoreFile), s"server$brokerId")
// set listener-specific configs and set an invalid path for the global config to verify that the overrides work
Seq(SecureInternal, SecureExternal).foreach { listenerName =>
diff --git a/core/src/test/scala/unit/kafka/utils/TestUtils.scala b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
index 2842d239c01..c9349201968 100755
--- a/core/src/test/scala/unit/kafka/utils/TestUtils.scala
+++ b/core/src/test/scala/unit/kafka/utils/TestUtils.scala
@@ -43,7 +43,7 @@ import org.apache.kafka.common.internals.Topic
import org.apache.kafka.common.memory.MemoryPool
import org.apache.kafka.common.message.UpdateMetadataRequestData.UpdateMetadataPartitionState
import org.apache.kafka.common.metrics.Metrics
-import org.apache.kafka.common.network.{ClientInformation, ListenerName, Mode}
+import org.apache.kafka.common.network.{ClientInformation, ListenerName, ConnectionMode}
import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.common.record._
import org.apache.kafka.common.requests._
@@ -358,7 +358,7 @@ object TestUtils extends Logging {
props.put(ServerConfigs.BACKGROUND_THREADS_CONFIG, "2")
if (protocolAndPorts.exists { case (protocol, _) => usesSslTransportLayer(protocol) })
- props ++= sslConfigs(Mode.SERVER, false, trustStoreFile, s"server$nodeId")
+ props ++= sslConfigs(ConnectionMode.SERVER, false, trustStoreFile, s"server$nodeId")
if (protocolAndPorts.exists { case (protocol, _) => usesSaslAuthentication(protocol) })
props ++= JaasTestUtils.saslConfigs(saslProperties)
@@ -646,7 +646,7 @@ object TestUtils extends Logging {
/**
* Returns security configuration options for broker or clients
*
- * @param mode Client or server mode
+ * @param connectionMode Client or server mode
* @param securityProtocol Security protocol which indicates if SASL or SSL or both configs are included
* @param trustStoreFile Trust store file must be provided for SSL and SASL_SSL
* @param certAlias Alias of certificate in SSL key store
@@ -656,7 +656,7 @@ object TestUtils extends Logging {
* @param needsClientCert If not empty, a flag which indicates if client certificates are required. By default
* client certificates are generated only if securityProtocol is SSL (not for SASL_SSL).
*/
- def securityConfigs(mode: Mode,
+ def securityConfigs(connectionMode: ConnectionMode,
securityProtocol: SecurityProtocol,
trustStoreFile: Option[File],
certAlias: String,
@@ -667,7 +667,7 @@ object TestUtils extends Logging {
val props = new Properties
if (usesSslTransportLayer(securityProtocol)) {
val addClientCert = needsClientCert.getOrElse(securityProtocol == SecurityProtocol.SSL)
- props ++= sslConfigs(mode, addClientCert, trustStoreFile, certAlias, certCn, tlsProtocol)
+ props ++= sslConfigs(connectionMode, addClientCert, trustStoreFile, certAlias, certCn, tlsProtocol)
}
if (usesSaslAuthentication(securityProtocol))
@@ -679,7 +679,7 @@ object TestUtils extends Logging {
def producerSecurityConfigs(securityProtocol: SecurityProtocol,
trustStoreFile: Option[File],
saslProperties: Option[Properties]): Properties =
- securityConfigs(Mode.CLIENT, securityProtocol, trustStoreFile, "producer", SslCertificateCn, saslProperties)
+ securityConfigs(ConnectionMode.CLIENT, securityProtocol, trustStoreFile, "producer", SslCertificateCn, saslProperties)
/**
* Create a (new) producer with a few pre-configured properties.
@@ -727,10 +727,10 @@ object TestUtils extends Logging {
}
def consumerSecurityConfigs(securityProtocol: SecurityProtocol, trustStoreFile: Option[File], saslProperties: Option[Properties]): Properties =
- securityConfigs(Mode.CLIENT, securityProtocol, trustStoreFile, "consumer", SslCertificateCn, saslProperties)
+ securityConfigs(ConnectionMode.CLIENT, securityProtocol, trustStoreFile, "consumer", SslCertificateCn, saslProperties)
def adminClientSecurityConfigs(securityProtocol: SecurityProtocol, trustStoreFile: Option[File], saslProperties: Option[Properties]): Properties =
- securityConfigs(Mode.CLIENT, securityProtocol, trustStoreFile, "admin-client", SslCertificateCn, saslProperties)
+ securityConfigs(ConnectionMode.CLIENT, securityProtocol, trustStoreFile, "admin-client", SslCertificateCn, saslProperties)
/**
* Create a consumer with a few pre-configured properties.
@@ -1404,7 +1404,7 @@ object TestUtils extends Logging {
new String(bytes, encoding)
}
- def sslConfigs(mode: Mode, clientCert: Boolean, trustStoreFile: Option[File], certAlias: String,
+ def sslConfigs(mode: ConnectionMode, clientCert: Boolean, trustStoreFile: Option[File], certAlias: String,
certCn: String = SslCertificateCn,
tlsProtocol: String = TestSslUtils.DEFAULT_TLS_PROTOCOL_FOR_TESTS): Properties = {
val trustStore = trustStoreFile.getOrElse {
diff --git a/streams/src/test/java/org/apache/kafka/streams/integration/ResetIntegrationWithSslTest.java b/streams/src/test/java/org/apache/kafka/streams/integration/ResetIntegrationWithSslTest.java
index 1574098e7cb..65a99a0b1b5 100644
--- a/streams/src/test/java/org/apache/kafka/streams/integration/ResetIntegrationWithSslTest.java
+++ b/streams/src/test/java/org/apache/kafka/streams/integration/ResetIntegrationWithSslTest.java
@@ -16,7 +16,7 @@
*/
package org.apache.kafka.streams.integration;
-import org.apache.kafka.common.network.Mode;
+import org.apache.kafka.common.network.ConnectionMode;
import org.apache.kafka.network.SocketServerConfigs;
import org.apache.kafka.streams.integration.utils.EmbeddedKafkaCluster;
import org.apache.kafka.test.TestSslUtils;
@@ -53,7 +53,7 @@ public class ResetIntegrationWithSslTest extends AbstractResetIntegrationTest {
brokerProps.put(SocketServerConfigs.CONNECTIONS_MAX_IDLE_MS_CONFIG, -1L);
try {
- SSL_CONFIG = TestSslUtils.createSslConfig(false, true, Mode.SERVER, TestUtils.tempFile(), "testCert");
+ SSL_CONFIG = TestSslUtils.createSslConfig(false, true, ConnectionMode.SERVER, TestUtils.tempFile(), "testCert");
brokerProps.put(SocketServerConfigs.LISTENERS_CONFIG, "SSL://localhost:0");
brokerProps.put(INTER_BROKER_LISTENER_NAME_CONFIG, "SSL");