diff --git a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java index 4a728baffccae..cd3a4bfad8722 100644 --- a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java +++ b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java @@ -42,6 +42,7 @@ import org.apache.kafka.common.requests.MetadataResponse; import org.apache.kafka.common.requests.RequestHeader; import org.apache.kafka.common.requests.ResponseHeader; +import org.apache.kafka.common.security.authenticator.SaslClientAuthenticator; import org.apache.kafka.common.utils.LogContext; import org.apache.kafka.common.utils.Time; import org.apache.kafka.common.utils.Utils; @@ -938,9 +939,15 @@ private void handleInitiateApiVersionRequests(long now) { * Validate that the response corresponds to the request we expect or else explode */ private static void correlate(RequestHeader requestHeader, ResponseHeader responseHeader) { - if (requestHeader.correlationId() != responseHeader.correlationId()) + if (requestHeader.correlationId() != responseHeader.correlationId()) { + if (SaslClientAuthenticator.isReserved(requestHeader.correlationId()) + && !SaslClientAuthenticator.isReserved(responseHeader.correlationId())) + throw new SchemaException("the response is unrelated to Sasl request since its correlation id is " + responseHeader.correlationId() + + " and the reserved range for Sasl request is [ " + + SaslClientAuthenticator.MIN_RESERVED_CORRELATION_ID + "," + SaslClientAuthenticator.MAX_RESERVED_CORRELATION_ID + "]"); throw new IllegalStateException("Correlation id for response (" + responseHeader.correlationId() + ") does not match request (" + requestHeader.correlationId() + "), request header: " + requestHeader); + } } /** @@ -1155,6 +1162,15 @@ public ClientRequest newClientRequest(String nodeId, return newClientRequest(nodeId, requestBuilder, createdTimeMs, expectResponse, defaultRequestTimeoutMs, null); } + // visible for testing + int nextCorrelationId() { + if (SaslClientAuthenticator.isReserved(correlation)) { + // the numeric overflow is fine as negative values is acceptable + correlation = SaslClientAuthenticator.MAX_RESERVED_CORRELATION_ID + 1; + } + return correlation++; + } + @Override public ClientRequest newClientRequest(String nodeId, AbstractRequest.Builder requestBuilder, @@ -1162,7 +1178,7 @@ public ClientRequest newClientRequest(String nodeId, boolean expectResponse, int requestTimeoutMs, RequestCompletionHandler callback) { - return new ClientRequest(nodeId, requestBuilder, correlation++, clientId, createdTimeMs, expectResponse, + return new ClientRequest(nodeId, requestBuilder, nextCorrelationId(), clientId, createdTimeMs, expectResponse, requestTimeoutMs, callback); } diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java index e972da15f32c0..8b32e81451760 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java +++ b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslClientAuthenticator.java @@ -105,6 +105,35 @@ public enum SaslState { private static final short DISABLE_KAFKA_SASL_AUTHENTICATE_HEADER = -1; private static final Random RNG = new Random(); + /** + * the reserved range of correlation id for Sasl requests. + * + * Noted: there is a story about reserved range. The response of LIST_OFFSET is compatible to response of SASL_HANDSHAKE. + * Hence, we could miss the schema error when using schema of SASL_HANDSHAKE to parse response of LIST_OFFSET. + * For example: the IllegalStateException caused by mismatched correlation id is thrown if following steps happens. + * 1) sent LIST_OFFSET + * 2) sent SASL_HANDSHAKE + * 3) receive response of LIST_OFFSET + * 4) succeed to use schema of SASL_HANDSHAKE to parse response of LIST_OFFSET + * 5) throw IllegalStateException due to mismatched correlation id + * As a simple approach, we force Sasl requests to use a reserved correlation id which is separated from those + * used in NetworkClient for Kafka requests. Hence, we can guarantee that every SASL request will throw + * SchemaException due to correlation id mismatch during reauthentication + */ + public static final int MAX_RESERVED_CORRELATION_ID = Integer.MAX_VALUE; + + /** + * We only expect one request in-flight a time during authentication so the small range is fine. + */ + public static final int MIN_RESERVED_CORRELATION_ID = MAX_RESERVED_CORRELATION_ID - 7; + + /** + * @return true if the correlation id is reserved for SASL request. otherwise, false + */ + public static boolean isReserved(int correlationId) { + return correlationId >= MIN_RESERVED_CORRELATION_ID; + } + private final Subject subject; private final String servicePrincipal; private final String host; @@ -178,7 +207,8 @@ public SaslClientAuthenticator(Map configs, } } - private SaslClient createSaslClient() { + // visible for testing + SaslClient createSaslClient() { try { return Subject.doAs(subject, (PrivilegedExceptionAction) () -> { String[] mechs = {mechanism}; @@ -326,6 +356,13 @@ public Long reauthenticationLatencyMs() { return reauthInfo.reauthenticationLatencyMs(); } + // visible for testing + int nextCorrelationId() { + if (!isReserved(correlationId)) + correlationId = MIN_RESERVED_CORRELATION_ID; + return correlationId++; + } + private RequestHeader nextRequestHeader(ApiKeys apiKey, short version) { String clientId = (String) configs.get(CommonClientConfigs.CLIENT_ID_CONFIG); short requestApiKey = apiKey.id; @@ -334,7 +371,7 @@ private RequestHeader nextRequestHeader(ApiKeys apiKey, short version) { setRequestApiKey(requestApiKey). setRequestApiVersion(version). setClientId(clientId). - setCorrelationId(correlationId++), + setCorrelationId(nextCorrelationId()), apiKey.requestHeaderVersion(version)); return currentRequestHeader; } diff --git a/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java b/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java index ccb0fee66f2fc..c4bde51d2b296 100644 --- a/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java +++ b/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java @@ -38,6 +38,7 @@ import org.apache.kafka.common.requests.ProduceRequest; import org.apache.kafka.common.requests.RequestHeader; import org.apache.kafka.common.requests.ResponseHeader; +import org.apache.kafka.common.security.authenticator.SaslClientAuthenticator; import org.apache.kafka.common.utils.LogContext; import org.apache.kafka.common.utils.MockTime; import org.apache.kafka.test.DelayedReceive; @@ -53,6 +54,9 @@ import java.util.Collections; import java.util.List; import java.util.Optional; +import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.IntStream; import static org.apache.kafka.common.protocol.ApiKeys.PRODUCE; import static org.junit.Assert.assertEquals; @@ -862,6 +866,16 @@ public void testCallDisconnect() throws Exception { assertTrue(client.canConnect(node, time.milliseconds())); } + @Test + public void testCorrelationId() { + int count = 100; + Set ids = IntStream.range(0, count) + .mapToObj(i -> client.nextCorrelationId()) + .collect(Collectors.toSet()); + assertEquals(count, ids.size()); + ids.forEach(id -> assertTrue(id < SaslClientAuthenticator.MIN_RESERVED_CORRELATION_ID)); + } + private RequestHeader parseHeader(ByteBuffer buffer) { buffer.getInt(); // skip size return RequestHeader.parse(buffer.slice()); 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 6c7d7c83606a8..b9c74c80cbf0f 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 @@ -29,10 +29,14 @@ import java.util.HashMap; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Random; import java.util.Base64.Encoder; +import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.Function; +import java.util.stream.Collectors; +import java.util.stream.IntStream; import javax.security.auth.Subject; import javax.security.auth.callback.Callback; @@ -44,10 +48,12 @@ import javax.security.auth.login.AppConfigurationEntry; import javax.security.auth.login.LoginContext; import javax.security.auth.login.LoginException; +import javax.security.sasl.SaslClient; import javax.security.sasl.SaslException; import org.apache.kafka.clients.NetworkClient; import org.apache.kafka.common.KafkaException; +import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.config.SaslConfigs; import org.apache.kafka.common.config.internals.BrokerSecurityConfigs; import org.apache.kafka.common.config.types.Password; @@ -75,6 +81,8 @@ import org.apache.kafka.common.network.TransportLayer; import org.apache.kafka.common.protocol.ApiKeys; import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.protocol.types.SchemaException; +import org.apache.kafka.common.requests.ListOffsetResponse; import org.apache.kafka.common.security.auth.Login; import org.apache.kafka.common.security.auth.SecurityProtocol; import org.apache.kafka.common.requests.AbstractRequest; @@ -115,9 +123,11 @@ import org.apache.kafka.common.utils.Time; import org.apache.kafka.common.utils.Utils; import org.junit.After; +import org.junit.Assert; import org.junit.Before; import org.junit.Test; +import static org.apache.kafka.common.protocol.ApiKeys.LIST_OFFSETS; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; @@ -1537,6 +1547,48 @@ public void testCannotReauthenticateWithDifferentPrincipal() throws Exception { server.verifyReauthenticationMetrics(0, 1); } } + + @Test + public void testCorrelationId() { + SaslClientAuthenticator authenticator = new SaslClientAuthenticator( + Collections.emptyMap(), + null, + "node", + null, + null, + null, + "plain", + false, + null, + null, + new LogContext() + ) { + @Override + SaslClient createSaslClient() { + return null; + } + }; + int count = (SaslClientAuthenticator.MAX_RESERVED_CORRELATION_ID - SaslClientAuthenticator.MIN_RESERVED_CORRELATION_ID) * 2; + Set ids = IntStream.range(0, count) + .mapToObj(i -> authenticator.nextCorrelationId()) + .collect(Collectors.toSet()); + assertEquals(SaslClientAuthenticator.MAX_RESERVED_CORRELATION_ID - SaslClientAuthenticator.MIN_RESERVED_CORRELATION_ID + 1, ids.size()); + ids.forEach(id -> { + assertTrue(id >= SaslClientAuthenticator.MIN_RESERVED_CORRELATION_ID); + assertTrue(SaslClientAuthenticator.isReserved(id)); + }); + } + + @Test + public void testConvertListOffsetResponseToSaslHandshakeResponse() { + ListOffsetResponse response = new ListOffsetResponse(0, Collections.singletonMap(new TopicPartition("topic", 0), + new ListOffsetResponse.PartitionData(Errors.NONE, 0, 0, Optional.empty()))); + ByteBuffer buffer = response.serialize(ApiKeys.LIST_OFFSETS, LIST_OFFSETS.latestVersion(), 0); + final RequestHeader header0 = new RequestHeader(LIST_OFFSETS, LIST_OFFSETS.latestVersion(), "id", SaslClientAuthenticator.MIN_RESERVED_CORRELATION_ID); + Assert.assertThrows(SchemaException.class, () -> NetworkClient.parseResponse(buffer.duplicate(), header0)); + final RequestHeader header1 = new RequestHeader(LIST_OFFSETS, LIST_OFFSETS.latestVersion(), "id", 1); + Assert.assertThrows(IllegalStateException.class, () -> NetworkClient.parseResponse(buffer.duplicate(), header1)); + } /** * Re-authentication must fail if mechanism changes