Skip to content

Cache the UpdateMetadataRequest when using the latest inter broker protocol version - #75

Merged
gitlw merged 6 commits into
linkedin:2.3-li-1from
gitlw:gitlw-cache-UpdateMetadataRequest-2
Mar 21, 2020
Merged

Cache the UpdateMetadataRequest when using the latest inter broker protocol version#75
gitlw merged 6 commits into
linkedin:2.3-li-1from
gitlw:gitlw-cache-UpdateMetadataRequest-2

Conversation

@gitlw

@gitlw gitlw commented Mar 9, 2020 •
edited
Loading

Copy link
Copy Markdown

To reduce the memory footprint on the controller, this PR caches the body of the UpdateMetadataRequest such that all the requests sent to many brokers can share the same
underlying bytes array.

All the tests in the "core" subprojects are still passing after this PR.
To test this change, I added some additional logs to compare the behaviors before and after.
Before this change, when a controller is elected, it sends out UpdateMetadataRequest by allocating bytes for the header and body repeatedly:

2020/03/17 03:18:43.914 INFO [AbstractRequest] [Controller-28772-to-broker-27051-send-thread] [kafka-server] [] allocating 15 bytes for header, 3639063 bytes for body
...
...
2020/03/17 03:18:44.205 INFO [AbstractRequest] [Controller-28772-to-broker-11992-send-thread] [kafka-server] [] allocating 15 bytes for header, 3639063 bytes for body

After this change, for a given round of UpdateMetadataRequest, the controller would allocate bytes for the header many times, but only once for the body

2020/03/18 18:55:56.503 WARN [AbstractRequest] [Controller-28772-to-broker-27051-send-thread] [kafka-server] [] In caching code, allocating 15 bytes for header, 2322938 bytes for body
2020/03/18 18:55:56.504 WARN [AbstractRequest] [Controller-28772-to-broker-5700-send-thread] [kafka-server] [] In caching code, allocating 15 bytes for header, 
2020/03/18 18:55:56.504 WARN [AbstractRequest] [Controller-28772-to-broker-27104-send-thread] [kafka-server] [] In caching code, allocating 15 bytes for header, 
2020/03/18 18:55:56.504 WARN [AbstractRequest] [Controller-28772-to-broker-27119-send-thread] [kafka-server] [] In caching code, allocating 15 bytes for header, 
...
...
2020/03/18 18:55:56.509 WARN [AbstractRequest] [Controller-28772-to-broker-5742-send-thread] [kafka-server] [] In caching code, allocating 15 bytes for header,

Committer Checklist (excluded from commit message)

  • Verify design and implementation
  • Verify test coverage and CI build status
  • Verify documentation (including upgrade notes)

@gitlw
gitlw requested review from hzxa21 and radai-rosenblatt March 9, 2020 19:13
@gitlw gitlw mentioned this pull request Mar 9, 2020
Closed
3 tasks
@gitlw
gitlw force-pushed the gitlw-cache-UpdateMetadataRequest-2 branch from f7679f2 to 323f103 Compare March 9, 2020 19:39
hzxa21
hzxa21 reviewed Mar 9, 2020
View reviewed changes

@hzxa21 hzxa21 left a comment

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Looks good overall. Thanks for the patch. Left one comment about where the caching logic should be.

Comment thread clients/src/main/java/org/apache/kafka/common/requests/AbstractRequest.java Outdated
@gitlw
gitlw marked this pull request as ready for review March 18, 2020 19:30
radai-rosenblatt
radai-rosenblatt reviewed Mar 20, 2020
View reviewed changes
Comment thread clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java Outdated
radai-rosenblatt
radai-rosenblatt reviewed Mar 20, 2020
View reviewed changes
Comment thread clients/src/main/java/org/apache/kafka/common/requests/UpdateMetadataRequest.java Outdated
@gitlw
gitlw merged commit dc6b292 into linkedin:2.3-li-1 Mar 21, 2020
@gitlw
gitlw deleted the gitlw-cache-UpdateMetadataRequest-2 branch March 21, 2020 00:42
@gitlw gitlw mentioned this pull request May 5, 2020
Closed
gitlw added a commit that referenced this pull request Jun 12, 2020
@gitlw
[LI-HOTFIX] Cache the UpdateMetadataRequest when using the latest int…
0e43c13 
…er broker protocol version (#75)

TICKET = KAFKA-7186
LI_DESCERIPTION = Cache the UpdateMetadataRequest when using the latest inter broker protocol
version
EXIT_CRITERIA = KAFKA-7186
gitlw added a commit that referenced this pull request Jun 13, 2020
@gitlw
[LI-HOTFIX] Cache the UpdateMetadataRequest when using the latest int…
b6aeca0 
…er broker protocol version (#75)

TICKET = KAFKA-7186
LI_DESCERIPTION = Cache the UpdateMetadataRequest when using the latest inter broker protocol
version
EXIT_CRITERIA = KAFKA-7186
gitlw added a commit that referenced this pull request Jun 17, 2020
@gitlw
[LI-HOTFIX] Cache the UpdateMetadataRequest when the inter broker pro…
902289e 
…tocol version >= 2.3-IV2 (#75)

TICKET = KAFKA-7186
LI_DESCERIPTION = Cache the UpdateMetadataRequest when using the latest inter broker protocol
version
EXIT_CRITERIA = KAFKA-7186
gitlw added a commit that referenced this pull request Jun 17, 2020
@gitlw
[LI-HOTFIX] Cache the UpdateMetadataRequest when the inter broker pro…
8adc2d1 
…tocol version >= 2.3-IV2 (#75)

TICKET = KAFKA-7186
LI_DESCERIPTION = Cache the UpdateMetadataRequest when using the latest inter broker protocol
version
EXIT_CRITERIA = KAFKA-7186
@gitlw gitlw mentioned this pull request Feb 16, 2022
Merged
3 tasks
gitlw added a commit to gitlw/kafka that referenced this pull request Feb 16, 2022
@gitlw
[LI-HOTFIX] Cache the UpdateMetadataRequest.Builder when the IBP is >…
5407a17 
…= KAFKA_2_3_IV2

TICKET = KAFKA-7186
LI_DESCERIPTION =
The original PR for this change is linkedin#75.
Before this patch, the controller needs to allocate an individual UpdateMetadataRequest.Builder
for each broker. This may incur significant memory overhead if the controller has just started,
and is trying to send the cluster's full metadata to all brokers.

This PR tries to reduce the memory footprint by reusing the same UpdateMetadataRequest.Builder.
This is only achievable when all of the UMR have the same body payload, which means
the common UMR should use the maxBrokerEpoch field instead of individual broker epochs.

EXIT_CRITERIA = KAFKA-7186
@gitlw gitlw mentioned this pull request Feb 16, 2022
Merged
3 tasks
gitlw added a commit that referenced this pull request Feb 16, 2022
@gitlw
[LI-HOTFIX] Cache the UpdateMetadataRequest.Builder when the IBP is >…
307d46c 
…= KAFKA_2_3_IV2 (#288)

TICKET = KAFKA-7186
LI_DESCERIPTION =
The original PR for this change is #75.
Before this patch, the controller needs to allocate an individual UpdateMetadataRequest.Builder
for each broker. This may incur significant memory overhead if the controller has just started,
and is trying to send the cluster's full metadata to all brokers.

This PR tries to reduce the memory footprint by reusing the same UpdateMetadataRequest.Builder.
This is only achievable when all of the UMR have the same body payload, which means
the common UMR should use the maxBrokerEpoch field instead of individual broker epochs.

EXIT_CRITERIA = KAFKA-7186
gitlw added a commit to gitlw/kafka that referenced this pull request Feb 17, 2022
@gitlw
[LI-HOTFIX] Cache the body buffer of UpdateMetadataRequests
ac8c657 
TICKET = KAFKA-7186
LI_DESCERIPTION =
This is a rewrite of the original PR linkedin#75.
In kafka 3.0, structs are removed, and requests can be serialized directly into ByteBuffers.
In vanilla kafka, the header and body are serialized directly into a single ByteBuffer inside the SendBuilder.

However, to reduce the memory footprint of UMRs, we would still want the ByteBufferSends
destined to different brokers to share the same body ByteBuffer. To achieve that, this PR modifies
the Java code generator so that the generated UpdateMetadataRequestData has the following
implementation for its write method

`
    private final ReentrantLock bodyBufferLock = new ReentrantLock();
    private ByteBuffer bodyBuffer = null;
    @OverRide
    public void write(Writable _writable, ObjectSerializationCache _cache, short _version) {
        bodyBufferLock.lock();
        try{
            if (bodyBuffer == null) {
                ObjectSerializationCache serializationCache = new ObjectSerializationCache();
                MessageSizeAccumulator messageSize = new MessageSizeAccumulator();
                addSize(messageSize, serializationCache, _version);
                bodyBuffer = ByteBuffer.allocate(messageSize.totalSize());
                doWrite(new ByteBufferAccessor(bodyBuffer), _cache, _version);
                bodyBuffer.flip();
            }
        } finally {
            bodyBufferLock.unlock();
        }
        _writable.writeByteBuffer(bodyBuffer);
    }
`

All other types of requests would end up with a write implementation as follows, which is
essentially the same as the vanilla implementation.
`
    @OverRide
    public void write(Writable _writable, ObjectSerializationCache _cache, short _version) {
        doWrite(_writable, _cache, _version);
    }

`

EXIT_CRITERIA = KAFKA-7186
gitlw added a commit that referenced this pull request Feb 17, 2022
@gitlw
[LI-HOTFIX] Cache the body buffer of UpdateMetadataRequests (#287)
a07c125 
TICKET = KAFKA-7186
LI_DESCERIPTION =
This is a rewrite of the original PR #75.
In kafka 3.0, structs are removed, and requests can be serialized directly into ByteBuffers.
In vanilla kafka, the header and body are serialized directly into a single ByteBuffer inside the SendBuilder.

However, to reduce the memory footprint of UMRs, we would still want the ByteBufferSends
destined to different brokers to share the same body ByteBuffer. To achieve that, this PR modifies
the Java code generator so that the generated UpdateMetadataRequestData has the following
implementation for its write method

`
    private final ReentrantLock bodyBufferLock = new ReentrantLock();
    private ByteBuffer bodyBuffer = null;
    @OverRide
    public void write(Writable _writable, ObjectSerializationCache _cache, short _version) {
        bodyBufferLock.lock();
        try{
            if (bodyBuffer == null) {
                ObjectSerializationCache serializationCache = new ObjectSerializationCache();
                MessageSizeAccumulator messageSize = new MessageSizeAccumulator();
                addSize(messageSize, serializationCache, _version);
                bodyBuffer = ByteBuffer.allocate(messageSize.totalSize());
                doWrite(new ByteBufferAccessor(bodyBuffer), _cache, _version);
                bodyBuffer.flip();
            }
        } finally {
            bodyBufferLock.unlock();
        }
        _writable.writeByteBuffer(bodyBuffer);
    }
`

All other types of requests would end up with a write implementation as follows, which is
essentially the same as the vanilla implementation.
`
    @OverRide
    public void write(Writable _writable, ObjectSerializationCache _cache, short _version) {
        doWrite(_writable, _cache, _version);
    }

`

EXIT_CRITERIA = KAFKA-7186
lmr3796 pushed a commit to lmr3796/kafka that referenced this pull request Mar 25, 2022
@gitlw @lmr3796
[LI-HOTFIX] Cache the UpdateMetadataRequest.Builder when the IBP is >…
b4ce1a5 
…= KAFKA_2_3_IV2 (linkedin#288)

TICKET = KAFKA-7186
LI_DESCERIPTION =
The original PR for this change is linkedin#75.
Before this patch, the controller needs to allocate an individual UpdateMetadataRequest.Builder
for each broker. This may incur significant memory overhead if the controller has just started,
and is trying to send the cluster's full metadata to all brokers.

This PR tries to reduce the memory footprint by reusing the same UpdateMetadataRequest.Builder.
This is only achievable when all of the UMR have the same body payload, which means
the common UMR should use the maxBrokerEpoch field instead of individual broker epochs.

EXIT_CRITERIA = KAFKA-7186
lmr3796 pushed a commit to lmr3796/kafka that referenced this pull request Mar 25, 2022
@gitlw @lmr3796
[LI-HOTFIX] Cache the body buffer of UpdateMetadataRequests (linkedin…
57cc525 
…#287)

TICKET = KAFKA-7186
LI_DESCERIPTION =
This is a rewrite of the original PR linkedin#75.
In kafka 3.0, structs are removed, and requests can be serialized directly into ByteBuffers.
In vanilla kafka, the header and body are serialized directly into a single ByteBuffer inside the SendBuilder.

However, to reduce the memory footprint of UMRs, we would still want the ByteBufferSends
destined to different brokers to share the same body ByteBuffer. To achieve that, this PR modifies
the Java code generator so that the generated UpdateMetadataRequestData has the following
implementation for its write method

`
    private final ReentrantLock bodyBufferLock = new ReentrantLock();
    private ByteBuffer bodyBuffer = null;
    @OverRide
    public void write(Writable _writable, ObjectSerializationCache _cache, short _version) {
        bodyBufferLock.lock();
        try{
            if (bodyBuffer == null) {
                ObjectSerializationCache serializationCache = new ObjectSerializationCache();
                MessageSizeAccumulator messageSize = new MessageSizeAccumulator();
                addSize(messageSize, serializationCache, _version);
                bodyBuffer = ByteBuffer.allocate(messageSize.totalSize());
                doWrite(new ByteBufferAccessor(bodyBuffer), _cache, _version);
                bodyBuffer.flip();
            }
        } finally {
            bodyBufferLock.unlock();
        }
        _writable.writeByteBuffer(bodyBuffer);
    }
`

All other types of requests would end up with a write implementation as follows, which is
essentially the same as the vanilla implementation.
`
    @OverRide
    public void write(Writable _writable, ObjectSerializationCache _cache, short _version) {
        doWrite(_writable, _cache, _version);
    }

`

EXIT_CRITERIA = KAFKA-7186
lmr3796 pushed a commit to lmr3796/kafka that referenced this pull request Jun 2, 2022
@gitlw @lmr3796
[LI-HOTFIX] Cache the UpdateMetadataRequest.Builder when the IBP is >…
4777449 
…= KAFKA_2_3_IV2 (linkedin#288)

TICKET = KAFKA-7186
LI_DESCERIPTION =
The original PR for this change is linkedin#75.
Before this patch, the controller needs to allocate an individual UpdateMetadataRequest.Builder
for each broker. This may incur significant memory overhead if the controller has just started,
and is trying to send the cluster's full metadata to all brokers.

This PR tries to reduce the memory footprint by reusing the same UpdateMetadataRequest.Builder.
This is only achievable when all of the UMR have the same body payload, which means
the common UMR should use the maxBrokerEpoch field instead of individual broker epochs.

EXIT_CRITERIA = KAFKA-7186
lmr3796 pushed a commit to lmr3796/kafka that referenced this pull request Jun 2, 2022
@gitlw @lmr3796
[LI-HOTFIX] Cache the body buffer of UpdateMetadataRequests (linkedin…
693c518 
…#287)

TICKET = KAFKA-7186
LI_DESCERIPTION =
This is a rewrite of the original PR linkedin#75.
In kafka 3.0, structs are removed, and requests can be serialized directly into ByteBuffers.
In vanilla kafka, the header and body are serialized directly into a single ByteBuffer inside the SendBuilder.

However, to reduce the memory footprint of UMRs, we would still want the ByteBufferSends
destined to different brokers to share the same body ByteBuffer. To achieve that, this PR modifies
the Java code generator so that the generated UpdateMetadataRequestData has the following
implementation for its write method

`
    private final ReentrantLock bodyBufferLock = new ReentrantLock();
    private ByteBuffer bodyBuffer = null;
    @OverRide
    public void write(Writable _writable, ObjectSerializationCache _cache, short _version) {
        bodyBufferLock.lock();
        try{
            if (bodyBuffer == null) {
                ObjectSerializationCache serializationCache = new ObjectSerializationCache();
                MessageSizeAccumulator messageSize = new MessageSizeAccumulator();
                addSize(messageSize, serializationCache, _version);
                bodyBuffer = ByteBuffer.allocate(messageSize.totalSize());
                doWrite(new ByteBufferAccessor(bodyBuffer), _cache, _version);
                bodyBuffer.flip();
            }
        } finally {
            bodyBufferLock.unlock();
        }
        _writable.writeByteBuffer(bodyBuffer);
    }
`

All other types of requests would end up with a write implementation as follows, which is
essentially the same as the vanilla implementation.
`
    @OverRide
    public void write(Writable _writable, ObjectSerializationCache _cache, short _version) {
        doWrite(_writable, _cache, _version);
    }

`

EXIT_CRITERIA = KAFKA-7186
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

2 more reviewers
@hzxa21 hzxa21 hzxa21 left review comments
@radai-rosenblatt radai-rosenblatt radai-rosenblatt approved these changes
Reviewers whose approvals may not affect merge requirements

Assignees

No one assigned

Labels

None yet

Projects

None yet

Milestone

No milestone

Development

Successfully merging this pull request may close these issues.

3 participants

@gitlw @radai-rosenblatt @hzxa21