diff --git a/LICENSE-binary b/LICENSE-binary index 057b88f0b2546..34c456ed0b9ba 100644 --- a/LICENSE-binary +++ b/LICENSE-binary @@ -219,16 +219,16 @@ jackson-module-jaxb-annotations-2.12.6 jackson-module-scala_2.13-2.12.6 jakarta.validation-api-2.0.2 javassist-3.27.0-GA -jetty-client-9.4.44.v20210927 -jetty-continuation-9.4.44.v20210927 -jetty-http-9.4.44.v20210927 -jetty-io-9.4.44.v20210927 -jetty-security-9.4.44.v20210927 -jetty-server-9.4.44.v20210927 -jetty-servlet-9.4.44.v20210927 -jetty-servlets-9.4.44.v20210927 -jetty-util-9.4.44.v20210927 -jetty-util-ajax-9.4.44.v20210927 +jetty-client-9.4.48.v20220622 +jetty-continuation-9.4.48.v20220622 +jetty-http-9.4.48.v20220622 +jetty-io-9.4.48.v20220622 +jetty-security-9.4.48.v20220622 +jetty-server-9.4.48.v20220622 +jetty-servlet-9.4.48.v20220622 +jetty-servlets-9.4.48.v20220622 +jetty-util-9.4.48.v20220622 +jetty-util-ajax-9.4.48.v20220622 jersey-common-2.34 jersey-server-2.34 log4j-1.2.17 diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml index b7a71927549a9..d47545432afbd 100644 --- a/checkstyle/suppressions.xml +++ b/checkstyle/suppressions.xml @@ -32,6 +32,8 @@ files="(Fetcher|Sender|SenderTest|ConsumerCoordinator|KafkaConsumer|KafkaProducer|Utils|TransactionManager|TransactionManagerTest|KafkaAdminClient|NetworkClient|Admin|KafkaRaftClient|KafkaRaftClientTest|RaftClientTestContext).java"/> + + + + diff --git a/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java b/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java index 0b90da8f800a6..8b7a9649c2c43 100644 --- a/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java +++ b/clients/src/main/java/org/apache/kafka/common/config/internals/BrokerSecurityConfigs.java @@ -36,6 +36,8 @@ public class BrokerSecurityConfigs { public static final String SASL_SERVER_CALLBACK_HANDLER_CLASS = "sasl.server.callback.handler.class"; public static final String SSL_PRINCIPAL_MAPPING_RULES_CONFIG = "ssl.principal.mapping.rules"; public static final String CONNECTIONS_MAX_REAUTH_MS = "connections.max.reauth.ms"; + public static final int DEFAULT_SASL_SERVER_MAX_RECEIVE_SIZE = 524288; + public static final String SASL_SERVER_MAX_RECEIVE_SIZE_CONFIG = "sasl.server.max.receive.size"; public static final String PRINCIPAL_BUILDER_CLASS_DOC = "The fully qualified name of a class that implements the " + "KafkaPrincipalBuilder interface, which is used to build the KafkaPrincipal object used during " + @@ -89,4 +91,8 @@ public class BrokerSecurityConfigs { + "The broker will disconnect any such connection that is not re-authenticated within the session lifetime and that is then subsequently " + "used for any purpose other than re-authentication. Configuration names can optionally be prefixed with listener prefix and SASL " + "mechanism name in lower-case. For example, listener.name.sasl_ssl.oauthbearer.connections.max.reauth.ms=3600000"; + + public static final String SASL_SERVER_MAX_RECEIVE_SIZE_DOC = "The maximum receive size allowed before and during initial SASL authentication." + + " Default receive size is 512KB. GSSAPI limits requests to 64K, but we allow upto 512KB by default for custom SASL mechanisms. In practice," + + " PLAIN, SCRAM and OAUTH mechanisms can use much smaller limits."; } diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/ByteBufferAccessor.java b/clients/src/main/java/org/apache/kafka/common/protocol/ByteBufferAccessor.java index bd0925d6db358..f643f5b5779b1 100644 --- a/clients/src/main/java/org/apache/kafka/common/protocol/ByteBufferAccessor.java +++ b/clients/src/main/java/org/apache/kafka/common/protocol/ByteBufferAccessor.java @@ -54,8 +54,15 @@ public double readDouble() { } @Override - public void readArray(byte[] arr) { + public byte[] readArray(int size) { + int remaining = buf.remaining(); + if (size > remaining) { + throw new RuntimeException("Error reading byte array of " + size + " byte(s): only " + remaining + + " byte(s) available"); + } + byte[] arr = new byte[size]; buf.get(arr); + return arr; } @Override diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/DataInputStreamReadable.java b/clients/src/main/java/org/apache/kafka/common/protocol/DataInputStreamReadable.java deleted file mode 100644 index 70ed52d6a02f6..0000000000000 --- a/clients/src/main/java/org/apache/kafka/common/protocol/DataInputStreamReadable.java +++ /dev/null @@ -1,139 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.kafka.common.protocol; - -import org.apache.kafka.common.utils.ByteUtils; - -import java.io.Closeable; -import java.io.DataInputStream; -import java.io.IOException; -import java.nio.ByteBuffer; - -public class DataInputStreamReadable implements Readable, Closeable { - protected final DataInputStream input; - - public DataInputStreamReadable(DataInputStream input) { - this.input = input; - } - - @Override - public byte readByte() { - try { - return input.readByte(); - } catch (IOException e) { - throw new RuntimeException(e); - } - } - - @Override - public short readShort() { - try { - return input.readShort(); - } catch (IOException e) { - throw new RuntimeException(e); - } - } - - @Override - public int readInt() { - try { - return input.readInt(); - } catch (IOException e) { - throw new RuntimeException(e); - } - } - - @Override - public long readLong() { - try { - return input.readLong(); - } catch (IOException e) { - throw new RuntimeException(e); - } - } - - @Override - public double readDouble() { - try { - return input.readDouble(); - } catch (IOException e) { - throw new RuntimeException(e); - } - } - - @Override - public void readArray(byte[] arr) { - try { - input.readFully(arr); - } catch (IOException e) { - throw new RuntimeException(e); - } - } - - @Override - public int readUnsignedVarint() { - try { - return ByteUtils.readUnsignedVarint(input); - } catch (IOException e) { - throw new RuntimeException(e); - } - } - - @Override - public ByteBuffer readByteBuffer(int length) { - byte[] arr = new byte[length]; - readArray(arr); - return ByteBuffer.wrap(arr); - } - - @Override - public int readVarint() { - try { - return ByteUtils.readVarint(input); - } catch (IOException e) { - throw new RuntimeException(e); - } - } - - @Override - public long readVarlong() { - try { - return ByteUtils.readVarlong(input); - } catch (IOException e) { - throw new RuntimeException(e); - } - } - - @Override - public int remaining() { - try { - return input.available(); - } catch (IOException e) { - throw new RuntimeException(e); - } - } - - @Override - public void close() { - try { - input.close(); - } catch (IOException e) { - throw new RuntimeException(e); - } - } - -} diff --git a/clients/src/main/java/org/apache/kafka/common/protocol/Readable.java b/clients/src/main/java/org/apache/kafka/common/protocol/Readable.java index 9c9e461ca806a..f453d12e1711c 100644 --- a/clients/src/main/java/org/apache/kafka/common/protocol/Readable.java +++ b/clients/src/main/java/org/apache/kafka/common/protocol/Readable.java @@ -32,7 +32,7 @@ public interface Readable { int readInt(); long readLong(); double readDouble(); - void readArray(byte[] arr); + byte[] readArray(int length); int readUnsignedVarint(); ByteBuffer readByteBuffer(int length); int readVarint(); @@ -40,8 +40,7 @@ public interface Readable { int remaining(); default String readString(int length) { - byte[] arr = new byte[length]; - readArray(arr); + byte[] arr = readArray(length); return new String(arr, StandardCharsets.UTF_8); } @@ -49,8 +48,7 @@ default List readUnknownTaggedField(List unknown if (unknowns == null) { unknowns = new ArrayList<>(); } - byte[] data = new byte[size]; - readArray(data); + byte[] data = readArray(size); unknowns.add(new RawTaggedField(tag, data)); return unknowns; } diff --git a/clients/src/main/java/org/apache/kafka/common/record/DefaultRecord.java b/clients/src/main/java/org/apache/kafka/common/record/DefaultRecord.java index 8772556b1dec1..b2235fef49039 100644 --- a/clients/src/main/java/org/apache/kafka/common/record/DefaultRecord.java +++ b/clients/src/main/java/org/apache/kafka/common/record/DefaultRecord.java @@ -342,6 +342,8 @@ private static DefaultRecord readFrom(ByteBuffer buffer, int numHeaders = ByteUtils.readVarint(buffer); if (numHeaders < 0) throw new InvalidRecordException("Found invalid number of record headers " + numHeaders); + if (numHeaders > buffer.remaining()) + throw new InvalidRecordException("Found invalid number of record headers. " + numHeaders + " is larger than the remaining size of the buffer"); final Header[] headers; if (numHeaders == 0) diff --git a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java index 6e35ee7a90ddb..a48d7472dcec7 100644 --- a/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java +++ b/clients/src/main/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticator.java @@ -27,6 +27,7 @@ import org.apache.kafka.common.errors.UnsupportedVersionException; import org.apache.kafka.common.message.SaslAuthenticateResponseData; import org.apache.kafka.common.message.SaslHandshakeResponseData; +import org.apache.kafka.common.network.InvalidReceiveException; import org.apache.kafka.common.network.Authenticator; import org.apache.kafka.common.network.ByteBufferSend; import org.apache.kafka.common.network.ChannelBuilders; @@ -88,8 +89,6 @@ import java.util.function.Supplier; public class SaslServerAuthenticator implements Authenticator { - // GSSAPI limits requests to 64K, but we allow a bit extra for custom SASL mechanisms - static final int MAX_RECEIVE_SIZE = 524288; private static final Logger LOG = LoggerFactory.getLogger(SaslServerAuthenticator.class); /** @@ -140,6 +139,7 @@ private enum SaslState { private String saslMechanism; // buffers used in `authenticate` + private Integer saslAuthRequestMaxReceiveSize; private NetworkReceive netInBuffer; private Send netOutBuffer; private Send authenticationFailureSend = null; @@ -189,6 +189,10 @@ public SaslServerAuthenticator(Map configs, // Note that the old principal builder does not support SASL, so we do not need to pass the // authenticator or the transport layer this.principalBuilder = ChannelBuilders.createPrincipalBuilder(configs, kerberosNameParser, null); + + saslAuthRequestMaxReceiveSize = (Integer) configs.get(BrokerSecurityConfigs.SASL_SERVER_MAX_RECEIVE_SIZE_CONFIG); + if (saslAuthRequestMaxReceiveSize == null) + saslAuthRequestMaxReceiveSize = BrokerSecurityConfigs.DEFAULT_SASL_SERVER_MAX_RECEIVE_SIZE; } private void createSaslServer(String mechanism) throws IOException { @@ -252,9 +256,13 @@ public void authenticate() throws IOException { } // allocate on heap (as opposed to any socket server memory pool) - if (netInBuffer == null) netInBuffer = new NetworkReceive(MAX_RECEIVE_SIZE, connectionId); + if (netInBuffer == null) netInBuffer = new NetworkReceive(saslAuthRequestMaxReceiveSize, connectionId); - netInBuffer.readFrom(transportLayer); + try { + netInBuffer.readFrom(transportLayer); + } catch (InvalidReceiveException e) { + throw new SaslAuthenticationException("Failing SASL authentication due to invalid receive size", e); + } if (!netInBuffer.complete()) return; netInBuffer.payload().rewind(); diff --git a/clients/src/test/java/org/apache/kafka/common/message/SimpleArraysMessageTest.java b/clients/src/test/java/org/apache/kafka/common/message/SimpleArraysMessageTest.java new file mode 100644 index 0000000000000..1b78adbb96227 --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/common/message/SimpleArraysMessageTest.java @@ -0,0 +1,54 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.common.message; + +import org.apache.kafka.common.protocol.ByteBufferAccessor; +import org.junit.jupiter.api.Test; + +import java.nio.ByteBuffer; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +public class SimpleArraysMessageTest { + @Test + public void testArrayBoundsChecking() { + // SimpleArraysMessageData takes 2 arrays + final ByteBuffer buf = ByteBuffer.wrap(new byte[] { + (byte) 0x7f, // Set size of first array to 126 which is larger than the size of this buffer + (byte) 0x00, (byte) 0x00, (byte) 0x00, (byte) 0x00, (byte) 0x00, (byte) 0x00, (byte) 0x00 + }); + final SimpleArraysMessageData out = new SimpleArraysMessageData(); + ByteBufferAccessor accessor = new ByteBufferAccessor(buf); + assertEquals("Tried to allocate a collection of size 126, but there are only 7 bytes remaining.", + assertThrows(RuntimeException.class, () -> out.read(accessor, (short) 2)).getMessage()); + } + + @Test + public void testArrayBoundsCheckingOtherArray() { + // SimpleArraysMessageData takes 2 arrays + final ByteBuffer buf = ByteBuffer.wrap(new byte[] { + (byte) 0x01, // Set size of first array to 0 + (byte) 0x7e, // Set size of second array to 125 which is larger than the size of this buffer + (byte) 0x00, (byte) 0x00, (byte) 0x00, (byte) 0x00, (byte) 0x00, (byte) 0x00 + }); + final SimpleArraysMessageData out = new SimpleArraysMessageData(); + ByteBufferAccessor accessor = new ByteBufferAccessor(buf); + assertEquals("Tried to allocate a collection of size 125, but there are only 6 bytes remaining.", + assertThrows(RuntimeException.class, () -> out.read(accessor, (short) 2)).getMessage()); + } +} diff --git a/clients/src/test/java/org/apache/kafka/common/protocol/ByteBufferAccessorTest.java b/clients/src/test/java/org/apache/kafka/common/protocol/ByteBufferAccessorTest.java new file mode 100644 index 0000000000000..6a0c6c2681c21 --- /dev/null +++ b/clients/src/test/java/org/apache/kafka/common/protocol/ByteBufferAccessorTest.java @@ -0,0 +1,58 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.common.protocol; + +import org.junit.jupiter.api.Test; + +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; + +public class ByteBufferAccessorTest { + @Test + public void testReadArray() { + ByteBuffer buf = ByteBuffer.allocate(1024); + ByteBufferAccessor accessor = new ByteBufferAccessor(buf); + final byte[] testArray = new byte[] {0x4b, 0x61, 0x46}; + accessor.writeByteArray(testArray); + accessor.writeInt(12345); + accessor.flip(); + final byte[] testArray2 = accessor.readArray(3); + assertArrayEquals(testArray, testArray2); + assertEquals(12345, accessor.readInt()); + assertEquals("Error reading byte array of 3 byte(s): only 0 byte(s) available", + assertThrows(RuntimeException.class, + () -> accessor.readArray(3)).getMessage()); + } + + @Test + public void testReadString() { + ByteBuffer buf = ByteBuffer.allocate(1024); + ByteBufferAccessor accessor = new ByteBufferAccessor(buf); + String testString = "ABC"; + final byte[] testArray = testString.getBytes(StandardCharsets.UTF_8); + accessor.writeByteArray(testArray); + accessor.flip(); + assertEquals("ABC", accessor.readString(3)); + assertEquals("Error reading byte array of 2 byte(s): only 0 byte(s) available", + assertThrows(RuntimeException.class, + () -> accessor.readString(2)).getMessage()); + } +} diff --git a/clients/src/test/java/org/apache/kafka/common/record/DefaultRecordTest.java b/clients/src/test/java/org/apache/kafka/common/record/DefaultRecordTest.java index 49743d2320135..67212165fc35a 100644 --- a/clients/src/test/java/org/apache/kafka/common/record/DefaultRecordTest.java +++ b/clients/src/test/java/org/apache/kafka/common/record/DefaultRecordTest.java @@ -247,6 +247,20 @@ public void testInvalidNumHeaders() { buf.flip(); assertThrows(InvalidRecordException.class, () -> DefaultRecord.readFrom(buf, 0L, 0L, RecordBatch.NO_SEQUENCE, null)); + + ByteBuffer buf2 = ByteBuffer.allocate(sizeOfBodyInBytes + ByteUtils.sizeOfVarint(sizeOfBodyInBytes)); + ByteUtils.writeVarint(sizeOfBodyInBytes, buf2); + buf2.put(attributes); + ByteUtils.writeVarlong(timestampDelta, buf2); + ByteUtils.writeVarint(offsetDelta, buf2); + ByteUtils.writeVarint(-1, buf2); // null key + ByteUtils.writeVarint(-1, buf2); // null value + ByteUtils.writeVarint(sizeOfBodyInBytes, buf2); // more headers than remaining buffer size, not allowed + buf2.position(buf2.limit()); + + buf2.flip(); + assertThrows(InvalidRecordException.class, + () -> DefaultRecord.readFrom(buf2, 0L, 0L, RecordBatch.NO_SEQUENCE, null)); } @Test diff --git a/clients/src/test/java/org/apache/kafka/common/requests/RequestContextTest.java b/clients/src/test/java/org/apache/kafka/common/requests/RequestContextTest.java index 4415ff960aafb..254dea0430ece 100644 --- a/clients/src/test/java/org/apache/kafka/common/requests/RequestContextTest.java +++ b/clients/src/test/java/org/apache/kafka/common/requests/RequestContextTest.java @@ -16,22 +16,31 @@ */ package org.apache.kafka.common.requests; +import org.apache.kafka.common.errors.InvalidRequestException; import org.apache.kafka.common.message.ApiVersionsResponseData; import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersionCollection; import org.apache.kafka.common.message.CreateTopicsResponseData; +import org.apache.kafka.common.message.ProduceRequestData; +import org.apache.kafka.common.message.SaslAuthenticateRequestData; import org.apache.kafka.common.network.ClientInformation; import org.apache.kafka.common.network.ListenerName; import org.apache.kafka.common.network.Send; import org.apache.kafka.common.protocol.ApiKeys; +import org.apache.kafka.common.protocol.ApiMessage; +import org.apache.kafka.common.protocol.ByteBufferAccessor; import org.apache.kafka.common.protocol.Errors; +import org.apache.kafka.common.protocol.ObjectSerializationCache; import org.apache.kafka.common.security.auth.KafkaPrincipal; import org.apache.kafka.common.security.auth.SecurityProtocol; import org.junit.jupiter.api.Test; import java.net.InetAddress; +import java.net.UnknownHostException; import java.nio.ByteBuffer; +import java.util.Collections; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; public class RequestContextTest { @@ -104,4 +113,78 @@ KafkaPrincipal.ANONYMOUS, new ListenerName("ssl"), SecurityProtocol.SASL_SSL, assertEquals(expectedResponse, parsedResponse.data()); } + @Test + public void testInvalidRequestForImplicitHashCollection() throws UnknownHostException { + short version = (short) 5; // choose a version with fixed length encoding, for simplicity + ByteBuffer corruptBuffer = produceRequest(version); + // corrupt the length of the topics array + corruptBuffer.putInt(8, (Integer.MAX_VALUE - 1) / 2); + + RequestHeader header = new RequestHeader(ApiKeys.PRODUCE, version, "console-producer", 3); + RequestContext context = new RequestContext(header, "0", InetAddress.getLocalHost(), + KafkaPrincipal.ANONYMOUS, new ListenerName("ssl"), SecurityProtocol.SASL_SSL, + ClientInformation.EMPTY, true); + + String msg = assertThrows(InvalidRequestException.class, + () -> context.parseRequest(corruptBuffer)).getCause().getMessage(); + assertEquals("Tried to allocate a collection of size 1073741823, but there are only 17 bytes remaining.", msg); + } + + @Test + public void testInvalidRequestForArrayList() throws UnknownHostException { + short version = (short) 5; // choose a version with fixed length encoding, for simplicity + ByteBuffer corruptBuffer = produceRequest(version); + // corrupt the length of the partitions array + corruptBuffer.putInt(17, Integer.MAX_VALUE); + + RequestHeader header = new RequestHeader(ApiKeys.PRODUCE, version, "console-producer", 3); + RequestContext context = new RequestContext(header, "0", InetAddress.getLocalHost(), + KafkaPrincipal.ANONYMOUS, new ListenerName("ssl"), SecurityProtocol.SASL_SSL, + ClientInformation.EMPTY, true); + + String msg = assertThrows(InvalidRequestException.class, + () -> context.parseRequest(corruptBuffer)).getCause().getMessage(); + assertEquals( + "Tried to allocate a collection of size 2147483647, but there are only 8 bytes remaining.", msg); + } + + private ByteBuffer produceRequest(short version) { + ProduceRequestData data = new ProduceRequestData() + .setAcks((short) -1) + .setTimeoutMs(1); + data.topicData().add( + new ProduceRequestData.TopicProduceData() + .setName("foo") + .setPartitionData(Collections.singletonList(new ProduceRequestData.PartitionProduceData() + .setIndex(42)))); + + return serialize(version, data); + } + + private ByteBuffer serialize(short version, ApiMessage data) { + ObjectSerializationCache cache = new ObjectSerializationCache(); + data.size(cache, version); + ByteBuffer buffer = ByteBuffer.allocate(1024); + data.write(new ByteBufferAccessor(buffer), cache, version); + buffer.flip(); + return buffer; + } + + @Test + public void testInvalidRequestForByteArray() throws UnknownHostException { + short version = (short) 1; // choose a version with fixed length encoding, for simplicity + ByteBuffer corruptBuffer = serialize(version, new SaslAuthenticateRequestData().setAuthBytes(new byte[0])); + // corrupt the length of the bytes array + corruptBuffer.putInt(0, Integer.MAX_VALUE); + + RequestHeader header = new RequestHeader(ApiKeys.SASL_AUTHENTICATE, version, "console-producer", 1); + RequestContext context = new RequestContext(header, "0", InetAddress.getLocalHost(), + KafkaPrincipal.ANONYMOUS, new ListenerName("ssl"), SecurityProtocol.SASL_SSL, + ClientInformation.EMPTY, true); + + String msg = assertThrows(InvalidRequestException.class, + () -> context.parseRequest(corruptBuffer)).getCause().getMessage(); + assertEquals("Error reading byte array of 2147483647 byte(s): only 0 byte(s) available", msg); + } + } diff --git a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java index 118a4244e2c49..679e5595dfc3e 100644 --- a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java +++ b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java @@ -184,6 +184,7 @@ import org.apache.kafka.common.protocol.ByteBufferAccessor; import org.apache.kafka.common.protocol.Errors; import org.apache.kafka.common.protocol.ObjectSerializationCache; +import org.apache.kafka.common.protocol.types.RawTaggedField; import org.apache.kafka.common.quota.ClientQuotaAlteration; import org.apache.kafka.common.quota.ClientQuotaEntity; import org.apache.kafka.common.quota.ClientQuotaFilter; @@ -205,10 +206,12 @@ import org.apache.kafka.common.security.token.delegation.TokenInformation; import org.apache.kafka.common.utils.SecurityUtils; import org.apache.kafka.common.utils.Utils; +import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import java.nio.BufferUnderflowException; import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -234,6 +237,7 @@ import static org.apache.kafka.common.protocol.ApiKeys.LIST_GROUPS; import static org.apache.kafka.common.protocol.ApiKeys.LIST_OFFSETS; import static org.apache.kafka.common.protocol.ApiKeys.OFFSET_FETCH; +import static org.apache.kafka.common.protocol.ApiKeys.SASL_AUTHENTICATE; import static org.apache.kafka.common.protocol.ApiKeys.STOP_REPLICA; import static org.apache.kafka.common.protocol.ApiKeys.SYNC_GROUP; import static org.apache.kafka.common.requests.FetchMetadata.INVALID_SESSION_ID; @@ -3047,4 +3051,92 @@ private ListTransactionsResponse createListTransactionsResponse() { return new ListTransactionsResponse(response); } + @Test + public void testInvalidSaslHandShakeRequest() { + AbstractRequest request = new SaslHandshakeRequest.Builder( + new SaslHandshakeRequestData().setMechanism("PLAIN")).build(); + ByteBuffer serializedBytes = request.serialize(); + // corrupt the length of the sasl mechanism string + serializedBytes.putShort(0, Short.MAX_VALUE); + + String msg = assertThrows(RuntimeException.class, () -> AbstractRequest. + parseRequest(request.apiKey(), request.version(), serializedBytes)).getMessage(); + assertEquals("Error reading byte array of 32767 byte(s): only 5 byte(s) available", msg); + } + + @Test + public void testInvalidSaslAuthenticateRequest() { + short version = (short) 1; // choose a version with fixed length encoding, for simplicity + byte[] b = new byte[] { + 0x11, 0x1f, 0x15, 0x2c, + 0x5e, 0x2a, 0x20, 0x26, + 0x6c, 0x39, 0x45, 0x1f, + 0x25, 0x1c, 0x2d, 0x25, + 0x43, 0x2a, 0x11, 0x76 + }; + SaslAuthenticateRequestData data = new SaslAuthenticateRequestData().setAuthBytes(b); + AbstractRequest request = new SaslAuthenticateRequest(data, version); + ByteBuffer serializedBytes = request.serialize(); + + // corrupt the length of the bytes array + serializedBytes.putInt(0, Integer.MAX_VALUE); + + String msg = assertThrows(RuntimeException.class, () -> AbstractRequest. + parseRequest(request.apiKey(), request.version(), serializedBytes)).getMessage(); + assertEquals("Error reading byte array of 2147483647 byte(s): only 20 byte(s) available", msg); + } + + @Test + public void testValidTaggedFieldsWithSaslAuthenticateRequest() { + byte[] byteArray = new byte[11]; + ByteBufferAccessor accessor = new ByteBufferAccessor(ByteBuffer.wrap(byteArray)); + + //construct a SASL_AUTHENTICATE request + byte[] authBytes = "test".getBytes(StandardCharsets.UTF_8); + accessor.writeUnsignedVarint(authBytes.length + 1); + accessor.writeByteArray(authBytes); + + //write total numbers of tags + accessor.writeUnsignedVarint(1); + + //write first tag + RawTaggedField taggedField = new RawTaggedField(1, new byte[] {0x1, 0x2, 0x3}); + accessor.writeUnsignedVarint(taggedField.tag()); + accessor.writeUnsignedVarint(taggedField.size()); + accessor.writeByteArray(taggedField.data()); + + accessor.flip(); + + SaslAuthenticateRequest saslAuthenticateRequest = (SaslAuthenticateRequest) AbstractRequest. + parseRequest(SASL_AUTHENTICATE, SASL_AUTHENTICATE.latestVersion(), accessor.buffer()).request; + Assertions.assertArrayEquals(authBytes, saslAuthenticateRequest.data().authBytes()); + assertEquals(1, saslAuthenticateRequest.data().unknownTaggedFields().size()); + assertEquals(taggedField, saslAuthenticateRequest.data().unknownTaggedFields().get(0)); + } + + @Test + public void testInvalidTaggedFieldsWithSaslAuthenticateRequest() { + byte[] byteArray = new byte[13]; + ByteBufferAccessor accessor = new ByteBufferAccessor(ByteBuffer.wrap(byteArray)); + + //construct a SASL_AUTHENTICATE request + byte[] authBytes = "test".getBytes(StandardCharsets.UTF_8); + accessor.writeUnsignedVarint(authBytes.length + 1); + accessor.writeByteArray(authBytes); + + //write total numbers of tags + accessor.writeUnsignedVarint(1); + + //write first tag + RawTaggedField taggedField = new RawTaggedField(1, new byte[] {0x1, 0x2, 0x3}); + accessor.writeUnsignedVarint(taggedField.tag()); + accessor.writeUnsignedVarint(Short.MAX_VALUE); // set wrong size for tagged field + accessor.writeByteArray(taggedField.data()); + + accessor.flip(); + + String msg = assertThrows(RuntimeException.class, () -> AbstractRequest. + parseRequest(SASL_AUTHENTICATE, SASL_AUTHENTICATE.latestVersion(), accessor.buffer())).getMessage(); + assertEquals("Error reading byte array of 32767 byte(s): only 3 byte(s) available", msg); + } } diff --git a/clients/src/test/java/org/apache/kafka/common/security/TestSecurityConfig.java b/clients/src/test/java/org/apache/kafka/common/security/TestSecurityConfig.java index 07cbb7856dded..197151f5fbc20 100644 --- a/clients/src/test/java/org/apache/kafka/common/security/TestSecurityConfig.java +++ b/clients/src/test/java/org/apache/kafka/common/security/TestSecurityConfig.java @@ -38,6 +38,8 @@ public class TestSecurityConfig extends AbstractConfig { null, Importance.MEDIUM, BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_DOC) .define(BrokerSecurityConfigs.CONNECTIONS_MAX_REAUTH_MS, Type.LONG, 0L, Importance.MEDIUM, BrokerSecurityConfigs.CONNECTIONS_MAX_REAUTH_MS_DOC) + .define(BrokerSecurityConfigs.SASL_SERVER_MAX_RECEIVE_SIZE_CONFIG, Type.INT, BrokerSecurityConfigs.DEFAULT_SASL_SERVER_MAX_RECEIVE_SIZE, + Importance.LOW, BrokerSecurityConfigs.SASL_SERVER_MAX_RECEIVE_SIZE_DOC) .withClientSslSupport() .withClientSaslSupport(); 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 988a0f2823213..40a27935f3f74 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 @@ -212,6 +212,52 @@ public void testValidSaslPlainOverPlaintext() throws Exception { checkAuthenticationAndReauthentication(securityProtocol, node); } + /** + * Test SASL/PLAIN with sasl.authentication.max.receive.size config + */ + @Test + public void testSaslAuthenticationMaxReceiveSize() throws Exception { + SecurityProtocol securityProtocol = SecurityProtocol.SASL_PLAINTEXT; + configureMechanisms("PLAIN", Collections.singletonList("PLAIN")); + + // test auth with 1KB receive size + saslServerConfigs.put(BrokerSecurityConfigs.SASL_SERVER_MAX_RECEIVE_SIZE_CONFIG, "1024"); + server = createEchoServer(securityProtocol); + + // test valid sasl authentication + String node1 = "valid"; + checkAuthenticationAndReauthentication(securityProtocol, node1); + + // test with handshake request with large mechanism string + byte[] bytes = new byte[1024]; + new Random().nextBytes(bytes); + String mechanism = new String(bytes, StandardCharsets.UTF_8); + String node2 = "invalid1"; + createClientConnection(SecurityProtocol.PLAINTEXT, node2); + SaslHandshakeRequest handshakeRequest = buildSaslHandshakeRequest(mechanism, ApiKeys.SASL_HANDSHAKE.latestVersion()); + RequestHeader header = new RequestHeader(ApiKeys.SASL_HANDSHAKE, handshakeRequest.version(), "someclient", nextCorrelationId++); + NetworkSend send = new NetworkSend(node2, handshakeRequest.toSend(header)); + selector.send(send); + //we will get exception in server and connection gets closed. + NetworkTestUtils.waitForChannelClose(selector, node2, ChannelState.READY.state()); + selector.close(); + + String node3 = "invalid2"; + createClientConnection(SecurityProtocol.PLAINTEXT, node3); + sendHandshakeRequestReceiveResponse(node3, ApiKeys.SASL_HANDSHAKE.latestVersion()); + + // test with sasl authenticate request with large auth_byes string + String authString = "\u0000" + TestJaasConfig.USERNAME + "\u0000" + new String(bytes, StandardCharsets.UTF_8); + ByteBuffer authBuf = ByteBuffer.wrap(Utils.utf8(authString)); + SaslAuthenticateRequestData data = new SaslAuthenticateRequestData().setAuthBytes(authBuf.array()); + SaslAuthenticateRequest request = new SaslAuthenticateRequest.Builder(data).build(); + header = new RequestHeader(ApiKeys.SASL_AUTHENTICATE, request.version(), "someclient", nextCorrelationId++); + send = new NetworkSend(node3, request.toSend(header)); + selector.send(send); + NetworkTestUtils.waitForChannelClose(selector, node3, ChannelState.READY.state()); + server.verifyAuthenticationMetrics(1, 2); + } + /** * Tests that SASL/PLAIN clients with invalid password fail authentication. */ diff --git a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticatorTest.java b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticatorTest.java index af0fedd4f5ad9..8245f57516002 100644 --- a/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticatorTest.java +++ b/clients/src/test/java/org/apache/kafka/common/security/authenticator/SaslServerAuthenticatorTest.java @@ -19,11 +19,11 @@ import java.net.InetAddress; import org.apache.kafka.common.config.internals.BrokerSecurityConfigs; import org.apache.kafka.common.errors.IllegalSaslStateException; +import org.apache.kafka.common.errors.SaslAuthenticationException; import org.apache.kafka.common.message.ApiMessageType; import org.apache.kafka.common.network.ChannelMetadataRegistry; import org.apache.kafka.common.network.ClientInformation; import org.apache.kafka.common.network.DefaultChannelMetadataRegistry; -import org.apache.kafka.common.network.InvalidReceiveException; import org.apache.kafka.common.network.ListenerName; import org.apache.kafka.common.network.TransportLayer; import org.apache.kafka.common.protocol.ApiKeys; @@ -68,10 +68,10 @@ public void testOversizeRequest() throws IOException { SCRAM_SHA_256.mechanismName(), new DefaultChannelMetadataRegistry()); when(transportLayer.read(any(ByteBuffer.class))).then(invocation -> { - invocation.getArgument(0).putInt(SaslServerAuthenticator.MAX_RECEIVE_SIZE + 1); + invocation.getArgument(0).putInt(BrokerSecurityConfigs.DEFAULT_SASL_SERVER_MAX_RECEIVE_SIZE + 1); return 4; }); - assertThrows(InvalidReceiveException.class, authenticator::authenticate); + assertThrows(SaslAuthenticationException.class, authenticator::authenticate); verify(transportLayer).read(any(ByteBuffer.class)); } diff --git a/clients/src/test/resources/common/message/SimpleArraysMessage.json b/clients/src/test/resources/common/message/SimpleArraysMessage.json new file mode 100644 index 0000000000000..76dc283b6a747 --- /dev/null +++ b/clients/src/test/resources/common/message/SimpleArraysMessage.json @@ -0,0 +1,29 @@ +// Licensed to the Apache Software Foundation (ASF) under one or more +// contributor license agreements. See the NOTICE file distributed with +// this work for additional information regarding copyright ownership. +// The ASF licenses this file to You under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance with +// the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. +{ + "name": "SimpleArraysMessage", + "type": "header", + "validVersions": "0-2", + "flexibleVersions": "1+", + "fields": [ + { "name": "Goats", "type": "[]StructArray", "versions": "1+", + "fields": [ + { "name": "Color", "type": "int8", "versions": "1+"}, + { "name": "Name", "type": "string", "versions": "2+"} + ] + }, + { "name": "Sheep", "type": "[]int32", "versions": "0+" } + ] +} diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java index 225fa7ff1f9e8..4c86745f9e41a 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java @@ -406,7 +406,7 @@ private void awaitStopConnector(String connName, long timeout) { } if (!connector.awaitShutdown(timeout)) { - log.error("Connector ‘{}’ failed to properly shut down, has become unresponsive, and " + log.error("Connector '{}' failed to properly shut down, has become unresponsive, and " + "may be consuming external resources. Correct the configuration for " + "this connector or remove the connector. After fixing the connector, it " + "may be necessary to restart this worker to release any consumed " diff --git a/core/src/main/scala/kafka/log/LogCleaner.scala b/core/src/main/scala/kafka/log/LogCleaner.scala index 12099f0f2c136..97de1db57c2e6 100644 --- a/core/src/main/scala/kafka/log/LogCleaner.scala +++ b/core/src/main/scala/kafka/log/LogCleaner.scala @@ -678,6 +678,8 @@ private[log] class Cleaner(val id: Int, if (discardBatchRecords) // The batch is only retained to preserve producer sequence information; the records can be removed false + else if (batch.isControlBatch) + true else Cleaner.this.shouldRetainRecord(map, retainDeletesAndTxnMarkers, batch, record, stats) } diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala index c556d7ab813b4..35d6f5b90a673 100755 --- a/core/src/main/scala/kafka/server/KafkaConfig.scala +++ b/core/src/main/scala/kafka/server/KafkaConfig.scala @@ -253,6 +253,7 @@ object Defaults { /** ********* General Security configuration ***********/ val ConnectionsMaxReauthMsDefault = 0L + val DefaultServerMaxMaxReceiveSize = BrokerSecurityConfigs.DEFAULT_SASL_SERVER_MAX_RECEIVE_SIZE val DefaultPrincipalSerde = classOf[DefaultKafkaPrincipalBuilder] /** ********* Sasl configuration ***********/ @@ -549,6 +550,7 @@ object KafkaConfig { /** ******** Common Security Configuration *************/ val PrincipalBuilderClassProp = BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_CONFIG val ConnectionsMaxReauthMsProp = BrokerSecurityConfigs.CONNECTIONS_MAX_REAUTH_MS + val SaslServerMaxReceiveSizeProp = BrokerSecurityConfigs.SASL_SERVER_MAX_RECEIVE_SIZE_CONFIG val securityProviderClassProp = SecurityConfig.SECURITY_PROVIDERS_CONFIG /** ********* SSL Configuration ****************/ @@ -960,6 +962,7 @@ object KafkaConfig { /** ******** Common Security Configuration *************/ val PrincipalBuilderClassDoc = BrokerSecurityConfigs.PRINCIPAL_BUILDER_CLASS_DOC val ConnectionsMaxReauthMsDoc = BrokerSecurityConfigs.CONNECTIONS_MAX_REAUTH_MS_DOC + val SaslServerMaxReceiveSizeDoc = BrokerSecurityConfigs.SASL_SERVER_MAX_RECEIVE_SIZE_DOC val securityProviderClassDoc = SecurityConfig.SECURITY_PROVIDERS_DOC /** ********* SSL Configuration ****************/ @@ -1249,6 +1252,7 @@ object KafkaConfig { /** ********* General Security Configuration ****************/ .define(ConnectionsMaxReauthMsProp, LONG, Defaults.ConnectionsMaxReauthMsDefault, MEDIUM, ConnectionsMaxReauthMsDoc) + .define(SaslServerMaxReceiveSizeProp, INT, Defaults.DefaultServerMaxMaxReceiveSize, MEDIUM, SaslServerMaxReceiveSizeDoc) .define(securityProviderClassProp, STRING, null, LOW, securityProviderClassDoc) /** ********* SSL Configuration ****************/ diff --git a/core/src/main/scala/kafka/tools/TestRaftServer.scala b/core/src/main/scala/kafka/tools/TestRaftServer.scala index 509913885ed8b..c067f945c5e4f 100644 --- a/core/src/main/scala/kafka/tools/TestRaftServer.scala +++ b/core/src/main/scala/kafka/tools/TestRaftServer.scala @@ -299,11 +299,7 @@ object TestRaftServer extends Logging { out.writeByteArray(data) } - override def read(input: protocol.Readable, size: Int): Array[Byte] = { - val data = new Array[Byte](size) - input.readArray(data) - data - } + override def read(input: protocol.Readable, size: Int): Array[Byte] = input.readArray(size) } private class LatencyHistogram( diff --git a/core/src/test/scala/kafka/raft/KafkaMetadataLogTest.scala b/core/src/test/scala/kafka/raft/KafkaMetadataLogTest.scala index e02529b33ddea..dcbe39d8321d6 100644 --- a/core/src/test/scala/kafka/raft/KafkaMetadataLogTest.scala +++ b/core/src/test/scala/kafka/raft/KafkaMetadataLogTest.scala @@ -878,11 +878,7 @@ object KafkaMetadataLogTest { override def write(data: Array[Byte], serializationCache: ObjectSerializationCache, out: Writable): Unit = { out.writeByteArray(data) } - override def read(input: protocol.Readable, size: Int): Array[Byte] = { - val array = new Array[Byte](size) - input.readArray(array) - array - } + override def read(input: protocol.Readable, size: Int): Array[Byte] = input.readArray(size) } val DefaultMetadataLogConfig = MetadataLogConfig( diff --git a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala index 5b9423421934d..253bf5490c58c 100755 --- a/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogCleanerTest.scala @@ -1027,6 +1027,50 @@ class LogCleanerTest { assertEquals(List(3, 4, 5), offsetsInLog(log)) } + + @Test + def testCleaningWithKeysConflictingWithTxnMarkerKeys(): Unit = { + val cleaner = makeCleaner(10) + val logProps = new Properties() + logProps.put(LogConfig.SegmentBytesProp, 1024: java.lang.Integer) + val log = makeLog(config = LogConfig.fromProps(logConfig.originals, logProps)) + val leaderEpoch = 5 + val producerEpoch = 0.toShort + + // First we append one committed transaction + val producerId1 = 1L + val appendProducer = appendTransactionalAsLeader(log, producerId1, producerEpoch, leaderEpoch) + appendProducer(Seq(1)) + log.appendAsLeader(commitMarker(producerId1, producerEpoch), leaderEpoch, origin = AppendOrigin.Coordinator) + + // Now we append one transaction with a key which conflicts with the COMMIT marker appended above + def commitRecordKey(): ByteBuffer = { + val keySize = ControlRecordType.COMMIT.recordKey().sizeOf() + val key = ByteBuffer.allocate(keySize) + ControlRecordType.COMMIT.recordKey().writeTo(key) + key.flip() + key + } + + val producerId2 = 2L + val records = MemoryRecords.withTransactionalRecords( + CompressionType.NONE, + producerId2, + producerEpoch, + 0, + new SimpleRecord(time.milliseconds(), commitRecordKey(), ByteBuffer.wrap("foo".getBytes)) + ) + log.appendAsLeader(records, leaderEpoch, origin = AppendOrigin.Client) + log.appendAsLeader(commitMarker(producerId2, producerEpoch), leaderEpoch, origin = AppendOrigin.Coordinator) + log.roll() + assertEquals(List(0, 1, 2, 3), offsetsInLog(log)) + + // After cleaning, the marker should not be removed + cleaner.clean(LogToClean(new TopicPartition("test", 0), log, 0L, log.activeSegment.baseOffset)) + assertEquals(List(0, 1, 2, 3), lastOffsetsPerBatchInLog(log)) + assertEquals(List(0, 1, 2, 3), offsetsInLog(log)) + } + @Test def testPartialSegmentClean(): Unit = { // because loadFactor is 0.75, this means we can fit 1 message in the map @@ -1917,20 +1961,31 @@ class LogCleanerTest { partitionLeaderEpoch, new SimpleRecord(key.toString.getBytes, value.toString.getBytes)) } - private def appendTransactionalAsLeader(log: Log, - producerId: Long, - producerEpoch: Short, - leaderEpoch: Int = 0, - origin: AppendOrigin = AppendOrigin.Client): Seq[Int] => LogAppendInfo = { - appendIdempotentAsLeader(log, producerId, producerEpoch, isTransactional = true, origin = origin) + private def appendTransactionalAsLeader( + log: Log, + producerId: Long, + producerEpoch: Short, + leaderEpoch: Int = 0, + origin: AppendOrigin = AppendOrigin.Client + ): Seq[Int] => LogAppendInfo = { + appendIdempotentAsLeader( + log, + producerId, + producerEpoch, + isTransactional = true, + leaderEpoch = leaderEpoch, + origin = origin + ) } - private def appendIdempotentAsLeader(log: Log, - producerId: Long, - producerEpoch: Short, - isTransactional: Boolean = false, - leaderEpoch: Int = 0, - origin: AppendOrigin = AppendOrigin.Client): Seq[Int] => LogAppendInfo = { + private def appendIdempotentAsLeader( + log: Log, + producerId: Long, + producerEpoch: Short, + isTransactional: Boolean = false, + leaderEpoch: Int = 0, + origin: AppendOrigin = AppendOrigin.Client + ): Seq[Int] => LogAppendInfo = { var sequence = 0 keys: Seq[Int] => { val simpleRecords = keys.map { key => diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala index 2e38df00f11af..a736bf43575c7 100755 --- a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala +++ b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala @@ -791,6 +791,8 @@ class KafkaConfigTest { case KafkaConfig.KafkaMetricsReporterClassesProp => // ignore case KafkaConfig.KafkaMetricsPollingIntervalSecondsProp => //ignore + case KafkaConfig.SaslServerMaxReceiveSizeProp => assertPropertyInvalid(baseProperties, name, "not_a_number") + // Raft Quorum Configs case RaftConfig.QUORUM_VOTERS_CONFIG => // ignore string case RaftConfig.QUORUM_ELECTION_TIMEOUT_MS_CONFIG => assertPropertyInvalid(baseProperties, name, "not_a_number") diff --git a/docs/js/templateData.js b/docs/js/templateData.js index 05541483dd87e..ee42207f61d79 100644 --- a/docs/js/templateData.js +++ b/docs/js/templateData.js @@ -19,6 +19,6 @@ limitations under the License. var context={ "version": "30", "dotVersion": "3.0", - "fullDotVersion": "3.0.2-SNAPSHOT", + "fullDotVersion": "3.0.3-SNAPSHOT", "scalaVersion": "2.13" }; diff --git a/docs/upgrade.html b/docs/upgrade.html index 68971701f2e53..b39b07e3cf5dd 100644 --- a/docs/upgrade.html +++ b/docs/upgrade.html @@ -19,6 +19,61 @@