You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by mi...@apache.org on 2022/11/28 14:15:57 UTC
[kafka] branch trunk updated: KAFKA-14414: Remove unnecessary usage of ObjectSerializationCache (#12890)
This is an automated email from the ASF dual-hosted git repository.
mimaison 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 edfa894eb5e KAFKA-14414: Remove unnecessary usage of ObjectSerializationCache (#12890)
edfa894eb5e is described below
commit edfa894eb5e2c4ba9dd262452781c0786be3f29f
Author: Divij Vaidya <di...@amazon.com>
AuthorDate: Mon Nov 28 15:15:34 2022 +0100
KAFKA-14414: Remove unnecessary usage of ObjectSerializationCache (#12890)
Reviewers: Mickael Maison <mi...@gmail.com>
---
.../kafka/common/requests/RequestHeader.java | 53 ++++++++++++++++++----
.../kafka/common/requests/ResponseHeader.java | 39 ++++++++++++++--
.../kafka/common/requests/RequestHeaderTest.java | 29 ++++++++++++
.../main/scala/kafka/network/RequestChannel.scala | 4 +-
core/src/main/scala/kafka/server/KafkaApis.scala | 2 +-
5 files changed, 111 insertions(+), 16 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java b/clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java
index 30a8f70e3c0..88a328d6616 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java
@@ -24,13 +24,16 @@ import org.apache.kafka.common.protocol.ByteBufferAccessor;
import org.apache.kafka.common.protocol.ObjectSerializationCache;
import java.nio.ByteBuffer;
+import java.util.Objects;
/**
* The header for a request in the Kafka protocol
*/
public class RequestHeader implements AbstractRequestResponse {
+ private final static int SIZE_NOT_INITIALIZED = -1;
private final RequestHeaderData data;
private final short headerVersion;
+ private int size = SIZE_NOT_INITIALIZED;
public RequestHeader(ApiKeys requestApiKey, short requestVersion, String clientId, int correlationId) {
this(new RequestHeaderData().
@@ -70,12 +73,39 @@ public class RequestHeader implements AbstractRequestResponse {
return data;
}
- public void write(ByteBuffer buffer, ObjectSerializationCache serializationCache) {
+ // Visible for testing.
+ void write(ByteBuffer buffer, ObjectSerializationCache serializationCache) {
data.write(new ByteBufferAccessor(buffer), serializationCache, headerVersion);
}
- public int size(ObjectSerializationCache serializationCache) {
- return data.size(serializationCache, headerVersion);
+ /**
+ * Calculates the size of {@link RequestHeader} in bytes.
+ *
+ * This method to calculate size should be only when it is immediately followed by
+ * {@link #write(ByteBuffer, ObjectSerializationCache)} method call. In such cases, ObjectSerializationCache
+ * helps to avoid the serialization twice. In all other cases, {@link #size()} should be preferred instead.
+ *
+ * Calls to this method leads to calculation of size every time it is invoked. {@link #size()} should be preferred
+ * instead.
+ *
+ * Visible for testing.
+ */
+ int size(ObjectSerializationCache serializationCache) {
+ this.size = data.size(serializationCache, headerVersion);
+ return size;
+ }
+
+ /**
+ * Returns the size of {@link RequestHeader} in bytes.
+ *
+ * Calls to this method are idempotent and inexpensive since it returns the cached value of size after the first
+ * invocation.
+ */
+ public int size() {
+ if (this.size == SIZE_NOT_INITIALIZED) {
+ this.size = size(new ObjectSerializationCache());
+ }
+ return size;
}
public ResponseHeader toResponseHeader() {
@@ -88,18 +118,23 @@ public class RequestHeader implements AbstractRequestResponse {
// We derive the header version from the request api version, so we read that first.
// The request api version is part of `RequestHeaderData`, so we reset the buffer position after the read.
int position = buffer.position();
+ int requestHeaderSize = buffer.remaining();
apiKey = buffer.getShort();
short apiVersion = buffer.getShort();
short headerVersion = ApiKeys.forId(apiKey).requestHeaderVersion(apiVersion);
buffer.position(position);
- RequestHeaderData headerData = new RequestHeaderData(
- new ByteBufferAccessor(buffer), headerVersion);
+ RequestHeaderData headerData = new RequestHeaderData(new ByteBufferAccessor(buffer), headerVersion);
// Due to a quirk in the protocol, client ID is marked as nullable.
// However, we treat a null client ID as equivalent to an empty client ID.
if (headerData.clientId() == null) {
headerData.setClientId("");
}
- return new RequestHeader(headerData, headerVersion);
+ final RequestHeader header = new RequestHeader(headerData, headerVersion);
+ // Size of a buffer required to serialize the information in this header is already known and would not
+ // change since the RequestHeader object is immutable. Instead of computing it again whenever
+ // RequestHeader#size() is called, we choose to cache the size value when available.
+ header.size = requestHeaderSize;
+ return header;
} catch (UnsupportedVersionException e) {
throw new InvalidRequestException("Unknown API key " + apiKey, e);
} catch (Throwable ex) {
@@ -114,6 +149,7 @@ public class RequestHeader implements AbstractRequestResponse {
", apiVersion=" + apiVersion() +
", clientId=" + clientId() +
", correlationId=" + correlationId() +
+ ", headerVersion=" + headerVersion +
")";
}
@@ -122,11 +158,12 @@ public class RequestHeader implements AbstractRequestResponse {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
RequestHeader that = (RequestHeader) o;
- return this.data.equals(that.data);
+ return headerVersion == that.headerVersion &&
+ Objects.equals(data, that.data);
}
@Override
public int hashCode() {
- return this.data.hashCode();
+ return Objects.hash(data, headerVersion);
}
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/ResponseHeader.java b/clients/src/main/java/org/apache/kafka/common/requests/ResponseHeader.java
index d45e28dc1d8..321e960d8cb 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/ResponseHeader.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/ResponseHeader.java
@@ -27,8 +27,10 @@ import java.util.Objects;
* A response header in the kafka protocol.
*/
public class ResponseHeader implements AbstractRequestResponse {
+ private final static int SIZE_NOT_INITIALIZED = -1;
private final ResponseHeaderData data;
private final short headerVersion;
+ private int size = SIZE_NOT_INITIALIZED;
public ResponseHeader(int correlationId, short headerVersion) {
this(new ResponseHeaderData().setCorrelationId(correlationId), headerVersion);
@@ -39,10 +41,35 @@ public class ResponseHeader implements AbstractRequestResponse {
this.headerVersion = headerVersion;
}
- public int size(ObjectSerializationCache serializationCache) {
+ /**
+ * Calculates the size of {@link ResponseHeader} in bytes.
+ *
+ * This method to calculate size should be only when it is immediately followed by
+ * {@link #write(ByteBuffer, ObjectSerializationCache)} method call. In such cases, ObjectSerializationCache
+ * helps to avoid the serialization twice. In all other cases, {@link #size()} should be preferred instead.
+ *
+ * Calls to this method leads to calculation of size every time it is invoked. {@link #size()} should be preferred
+ * instead.
+ *
+ * Visible for testing.
+ */
+ int size(ObjectSerializationCache serializationCache) {
return data().size(serializationCache, headerVersion);
}
+ /**
+ * Returns the size of {@link ResponseHeader} in bytes.
+ *
+ * Calls to this method are idempotent and inexpensive since it returns the cached value of size after the first
+ * invocation.
+ */
+ public int size() {
+ if (this.size == SIZE_NOT_INITIALIZED) {
+ this.size = size(new ObjectSerializationCache());
+ }
+ return size;
+ }
+
public int correlationId() {
return this.data.correlationId();
}
@@ -55,7 +82,8 @@ public class ResponseHeader implements AbstractRequestResponse {
return data;
}
- public void write(ByteBuffer buffer, ObjectSerializationCache serializationCache) {
+ // visible for testing
+ void write(ByteBuffer buffer, ObjectSerializationCache serializationCache) {
data.write(new ByteBufferAccessor(buffer), serializationCache, headerVersion);
}
@@ -68,9 +96,10 @@ public class ResponseHeader implements AbstractRequestResponse {
}
public static ResponseHeader parse(ByteBuffer buffer, short headerVersion) {
- return new ResponseHeader(
- new ResponseHeaderData(new ByteBufferAccessor(buffer), headerVersion),
- headerVersion);
+ final ResponseHeader header = new ResponseHeader(
+ new ResponseHeaderData(new ByteBufferAccessor(buffer), headerVersion), headerVersion);
+ header.size = buffer.remaining();
+ return header;
}
@Override
diff --git a/clients/src/test/java/org/apache/kafka/common/requests/RequestHeaderTest.java b/clients/src/test/java/org/apache/kafka/common/requests/RequestHeaderTest.java
index e4f19956d46..ed3ccedad7e 100644
--- a/clients/src/test/java/org/apache/kafka/common/requests/RequestHeaderTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/requests/RequestHeaderTest.java
@@ -25,6 +25,9 @@ import org.junit.jupiter.api.Test;
import java.nio.ByteBuffer;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
public class RequestHeaderTest {
@@ -113,4 +116,30 @@ public class RequestHeaderTest {
assertEquals(ApiKeys.FIND_COORDINATOR, parsed.apiKey());
assertEquals((short) 10, parsed.apiVersion());
}
+
+ @Test
+ public void verifySizeMethodsReturnSameValue() {
+ // Create a dummy RequestHeaderData
+ RequestHeaderData headerData = new RequestHeaderData().
+ setClientId("hakuna-matata").
+ setCorrelationId(123).
+ setRequestApiKey(ApiKeys.FIND_COORDINATOR.id).
+ setRequestApiVersion((short) 10);
+
+ // Serialize RequestHeaderData to a buffer
+ ObjectSerializationCache serializationCache = new ObjectSerializationCache();
+ ByteBuffer buffer = ByteBuffer.allocate(headerData.size(serializationCache, (short) 2));
+ headerData.write(new ByteBufferAccessor(buffer), serializationCache, (short) 2);
+ buffer.flip();
+
+ // actual call to generate the RequestHeader from buffer containing RequestHeaderData
+ RequestHeader parsed = spy(RequestHeader.parse(buffer));
+
+ // verify that the result of cached value is same as actual calculation of size
+ assertEquals(parsed.size(), parsed.size(new ObjectSerializationCache()));
+
+ // verify that size(ObjectSerializationCache) is only called once, i.e. during assertEquals call. This validates
+ // that size() method does not calculate the size instead it uses the cached value
+ verify(parsed).size(any(ObjectSerializationCache.class));
+ }
}
diff --git a/core/src/main/scala/kafka/network/RequestChannel.scala b/core/src/main/scala/kafka/network/RequestChannel.scala
index eba5e5f929c..76bb25e9237 100644
--- a/core/src/main/scala/kafka/network/RequestChannel.scala
+++ b/core/src/main/scala/kafka/network/RequestChannel.scala
@@ -33,7 +33,7 @@ import org.apache.kafka.common.memory.MemoryPool
import org.apache.kafka.common.message.ApiMessageType.ListenerType
import org.apache.kafka.common.message.EnvelopeResponseData
import org.apache.kafka.common.network.Send
-import org.apache.kafka.common.protocol.{ApiKeys, Errors, ObjectSerializationCache}
+import org.apache.kafka.common.protocol.{ApiKeys, Errors}
import org.apache.kafka.common.requests._
import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.common.utils.{Sanitizer, Time}
@@ -110,7 +110,7 @@ object RequestChannel extends Logging {
def sizeOfBodyInBytes: Int = bodyAndSize.size
- def sizeInBytes: Int = header.size(new ObjectSerializationCache) + sizeOfBodyInBytes
+ def sizeInBytes: Int = header.size + sizeOfBodyInBytes
//most request types are parsed entirely into objects at this point. for those we can release the underlying buffer.
//some (like produce, or any time the schema contains fields of types BYTES or NULLABLE_BYTES) retain a reference
diff --git a/core/src/main/scala/kafka/server/KafkaApis.scala b/core/src/main/scala/kafka/server/KafkaApis.scala
index b43445b2247..d5621a7cce7 100644
--- a/core/src/main/scala/kafka/server/KafkaApis.scala
+++ b/core/src/main/scala/kafka/server/KafkaApis.scala
@@ -546,7 +546,6 @@ class KafkaApis(val requestChannel: RequestChannel,
*/
def handleProduceRequest(request: RequestChannel.Request, requestLocal: RequestLocal): Unit = {
val produceRequest = request.body[ProduceRequest]
- val requestSize = request.sizeInBytes
if (RequestUtils.hasTransactionalRecords(produceRequest)) {
val isAuthorizedTransactional = produceRequest.transactionalId != null &&
@@ -609,6 +608,7 @@ class KafkaApis(val requestChannel: RequestChannel,
// have been violated. If both quotas have been violated, use the max throttle time between the two quotas. Note
// that the request quota is not enforced if acks == 0.
val timeMs = time.milliseconds()
+ val requestSize = request.sizeInBytes
val bandwidthThrottleTimeMs = quotas.produce.maybeRecordAndGetThrottleTimeMs(request, requestSize, timeMs)
val requestThrottleTimeMs =
if (produceRequest.acks == 0) 0