MINOR: Remove connection id from Send and consolidate request/message utils by ijuma · Pull Request #9714 · apache/kafka

ijuma · 2020-12-08T20:30:17Z

Connection id is now only present in NetworkSend, which is now
the class used by Selector/NetworkClient/KafkaChannel (which
works well since NetworkReceive is the class used for
received data).

The previous NetworkSend was also responsible for adding a size
prefix. This logic is already present in SendBuilder, but for the
minority of cases where SendBuilder is not used (including
a number of tests), we now have ByteBufferSend.sizePrefixed().

With regards to the request/message utilities:

Renamed toByteBuffer/toBytes in MessageUtil to
toVersionPrefixedByteBuffer/toVersionPrefixedBytes for clarity.
Introduced new MessageUtil.toByteBuffer that does not include
the version as the prefix.
Renamed serializeBody in AbstractRequest/Response to
serialize for symmetry with parse.
Introduced RequestTestUtils and moved relevant methods from
TestUtils.
Moved serializeWithHeader methods that were only used in
tests to RequestTestUtils.
Deleted MessageTestUtil.

Finally, a couple of changes to simplify coding patterns:

Added flip() and buffer() to ByteBufferAccessor.
Added MessageSizeAccumulator.sizeExcludingZeroCopy.
Used lambdas instead of TestCondition.
Used Arrays.copyOf instead of System.arraycopy in MessageUtil.

Committer Checklist (excluded from commit message)

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

…tils

chia7712

@ijuma Thanks for this great cleanup. Please take a look at following minor comments. Thanks!

chia7712 · 2020-12-09T04:09:07Z

+
+        int headerSize = header.size(cache, headerVersion);
+        int messageSize = apiMessage.size(cache, apiVersion);
+        ByteBufferAccessor writable = new ByteBufferAccessor(ByteBuffer.allocate(4 + headerSize + messageSize));


Why it need 4 byte if it exclude size prefix?

Great catch. This was a copy and paste bug and we had no tests verifying this. I have fixed it and added a test.

chia7712 · 2020-12-09T04:18:45Z

@@ -46,8 +45,7 @@ public class MultiRecordsSend implements Send {
     * Construct a MultiRecordsSend for the given destination from a queue of Send objects. The queue will be


This comment need to be updated.

chia7712 · 2020-12-09T04:19:20Z

@@ -24,11 +24,6 @@
 */
 public interface Send {


please remove "destination" from the docs

chia7712 · 2020-12-09T04:26:20Z

    // `openOrClosingChannel` can be None if the selector closed the connection because it was idle for too long
    if (openOrClosingChannel(connectionId).isDefined) {
-      selector.send(responseSend)
+      selector.send(new NetworkSend(response.request.context.connectionId, responseSend))


line#955 already has connectionId.

chia7712 · 2020-12-09T04:30:04Z

-        handleChannelMuteEvent(send.destination, ChannelMuteEvent.RESPONSE_SENT)
-        tryUnmuteChannel(send.destination)
+        handleChannelMuteEvent(send.destinationId(), ChannelMuteEvent.RESPONSE_SENT)
+        tryUnmuteChannel(send.destinationId())


How about keeping "parameterless"? fewer changes and more clear.

Looks like the IDE added this when I renamed it or something. Fixed.

chia7712 · 2020-12-09T04:44:58Z

+                                                      final int numNodes,
+                                                      final Map<String, Errors> topicErrors,
+                                                      final Map<String, Integer> topicPartitionCounts,
+                                                      final short responseVersion) {


unused argument

Looks like a test bug from the previous PR. Will fix so that it's used.

chia7712 · 2020-12-09T04:51:44Z

+    public static ByteBuffer serializeRequestHeader(RequestHeader header) {
+        ObjectSerializationCache serializationCache = new ObjectSerializationCache();
+        ByteBuffer buffer = ByteBuffer.allocate(header.size(serializationCache));
+        header.write(buffer, serializationCache);


RequestHeader#write has only 2 usages and both of them are in test scope. It should be fine to remove RequestHeader#write from production.

It's about conceptual integrity for the class. It should provide a mechanism for serialization that doesn't require reaching into its internal structures.

fair enough

ijuma · 2020-12-09T05:44:18Z

@chia7712 Thanks for the review, I'm particularly glad you caught the 4 wasted bytes. :) I pushed updates and left a comment in the one case I haven't addressed.

chia7712

@ijuma overall +1 and thanks for this great refactor. please take a look at last comment :)

chia7712 · 2020-12-09T07:16:49Z

+/**
+ * A size delimited Send that consists of a 4 byte network-ordered size N followed by N bytes of content.
+ */
+public class SizeDelimitedSend extends ByteBufferSend {


How about making SizeDelimitedSend be a static method in ByteBufferSend? For example:

public static Send withSizeDelimited(ByteBuffer buffer) { ByteBuffer sizeBuffer = ByteBuffer.allocate(4); sizeBuffer.putInt(0, buffer.remaining()); return new ByteBufferSend(sizeBuffer, buffer); }

I had the same thought and initially did that, but it's an issue for many tests since they verify the properties of the ByteBuffer and it is not accessible if we do this.

but it's an issue for many tests since they verify the properties of the ByteBuffer and it is not accessible if we do this.

Can it be resolved if we return ByteBuffer rather than Send?

It has to be a Send too. I looked at this a bit closer and we can just return ByteBufferSend from the static factory method. The thing I attempted before was to combine at the NetworkSend level and that doesn't work since NetworkSend can work with Sends that are not backed by a ByteBuffer. But it actually works fine at ByteBufferSend level. Thanks for the suggestion, please check the update.

…tory method

chia7712 · 2020-12-09T14:27:05Z

1 to latest commit

hachikuji

Nice cleanup! I didn't look too carefully, but it seems like a step in the right direction. Pulling destination out of Send is something I've wanted to do for some time.

ijuma · 2020-12-09T19:14:29Z

Tests passed, merging to trunk.

ijuma force-pushed the remove-connection-id-from-send branch from 23ebe94 to 87db298 Compare December 8, 2020 20:32

MINOR: Remove connection id from Send and consolidate serialization u…

9b849c4

…tils

ijuma force-pushed the remove-connection-id-from-send branch from 87db298 to 9b849c4 Compare December 8, 2020 23:50

ijuma marked this pull request as ready for review December 9, 2020 00:39

ijuma changed the title ~~MINOR: Remove connection id from Send and consolidate serialization utils~~ MINOR: Remove connection id from Send and consolidate request/message utils Dec 9, 2020

ijuma requested a review from chia7712 December 9, 2020 03:49

Minor tweak

59c1fff

ijuma mentioned this pull request Dec 9, 2020

KAFKA-10818: Skip conversion to Struct when serializing generated requests/responses #7409

Merged

3 tasks

chia7712 reviewed Dec 9, 2020

View reviewed changes

ijuma added 5 commits December 8, 2020 21:38

Comment updates

13c7d5c

Fix 4 bytes wasted in RequestUtils.serialize

c1db89c

Remove unnecessary rewind calls

b164eca

Use responseVersion in RequestTestUtils.metadataResponse

742ea7d

Style fixes

f00dbe3

chia7712 approved these changes Dec 9, 2020

View reviewed changes

Replace SizeDelimitedSend with ByteBufferSend.sizePrefixed static fac…

e86ed57

…tory method

hachikuji reviewed Dec 9, 2020

View reviewed changes

ijuma merged commit 1f98112 into apache:trunk Dec 9, 2020

ijuma deleted the remove-connection-id-from-send branch December 9, 2020 19:16

		@@ -46,8 +45,7 @@ public class MultiRecordsSend implements Send {
		* Construct a MultiRecordsSend for the given destination from a queue of Send objects. The queue will be

Conversation

ijuma commented Dec 8, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Committer Checklist (excluded from commit message)

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ijuma Dec 9, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ijuma commented Dec 9, 2020

Uh oh!

chia7712 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ijuma Dec 9, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

chia7712 commented Dec 9, 2020

Uh oh!

hachikuji left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

ijuma commented Dec 9, 2020

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

ijuma commented Dec 8, 2020 •

edited

Loading

ijuma Dec 9, 2020 •

edited

Loading

ijuma Dec 9, 2020 •

edited

Loading

hachikuji left a comment •

edited

Loading