Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[compat][server][client][test] Global RT DIV improvement (part 2): Ch… #1257

Open
wants to merge 1 commit into
base: main
Choose a base branch
from

Conversation

lluwm
Copy link
Contributor

@lluwm lluwm commented Oct 23, 2024

…unking support for DIV message

This change mainly focuses on adding chunking support for DIV messages when they are produced to Kafka topics, as the size of DIV message can be large. We leverage today's chunking mechanism for regular records and extend it to support DIV with the following modifications:

  1. All the DIV messages are of type CONTROL_MESSAGE_DIV in its KafkaKey and their corresponding KafkaMessageEnvelope uses Put as the payload.
  2. Inside the Put payload, the actual message is stored in the putValue field and the schemaId can have the following cases:
    • If the DIV message is non-chunked, the schemaId is set to GLOBAL_DIV_STATE.
    • If the DIV message is chunk message, the schemaId is set to CHUNK.
    • If the DIV message is a chunk manifest message, the schemaId is set to CHUNKED_VALUE_MANIFEST.
  3. ChunkAssembler is adapted, on the receiver side, to buffer, assemble, and deserialize DIV messages (chunked/non-chunked).

How was this PR tested?

  • new integration test.
  • passed CI

Does this PR introduce any user-facing changes?

  • No. You can skip the rest of this section.
  • Yes. Make sure to explain your proposed changes and call out the behavior change.

…unking support for DIV message

This change mainly focuses on adding chunking support for DIV messages when they are produced to Kafka topics,
as the size of DIV message can be large. We leverage today's chunking mechanism for regular records and extend
it to support DIV with the following modifications:

1. All the DIV messages are of type {@link MessageType#CONTROL_MESSAGE_DIV} in its KafkaKey and their corresponding
   KafkaMessageEnvelope uses Put as the payload.
2. Inside the Put payload, the actual message is stored in the putValue field and the schemaId can have the following cases:
   - If the DIV message is non-chunked, the schemaId is set to GLOBAL_DIV_STATE.
   - If the DIV message is chunk message, the schemaId is set to CHUNK.
   - If the DIV message is a chunk manifest message, the schemaId is set to CHUNKED_VALUE_MANIFEST.
3. ChunkAssembler is adapted, on the receiver side, to buffer, assemble, and deserialize DIV messages (chunked/non-chunked).
@@ -237,6 +239,8 @@ public abstract class StoreIngestionTask implements Runnable, Closeable {
* flushed to the metadata partition of the storage engine regularly in {@link #syncOffset(String, PartitionConsumptionState)}
*/
private final KafkaDataIntegrityValidator kafkaDataIntegrityValidator;
private ChunkAssembler divChunkAssembler;

Choose a reason for hiding this comment

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

Making it final? Or it would be difficult to construct a test?

@@ -4336,4 +4384,9 @@ public boolean hasAllPartitionReportedCompleted() {
void setVersionRole(PartitionReplicaIngestionContext.VersionRole versionRole) {
this.versionRole = versionRole;
}

// Only for testing purpose.

Choose a reason for hiding this comment

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

I understand everyone has their own habit. May I suggest to still make the divChunkAssembler final but creating a constructor that allows you to pass a mock divChunkAseembler? In that way, you don't have to have such a setter.

Either way needs one more method but the "setting a variable final" is better in execution for performance. Just a bit.

Copy link
Contributor Author

@lluwm lluwm Oct 25, 2024

Choose a reason for hiding this comment

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

Thanks for the review, @lusong64! SG. I will fix it.

Copy link

@lusong64 lusong64 left a comment

Choose a reason for hiding this comment

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

Just some minor comments. No need to be a blocker.

Put put = new Put();
value.payloadUnion = put;
value.messageType = MessageType.PUT.getValue();
PubSubTopic verstionTopic = pubSubTopicRepository.getTopic(Version.composeKafkaTopic("testStore", 1));

Choose a reason for hiding this comment

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

typo? versionTopic?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

oh, yes. Will fix it.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Labels
None yet
Projects
None yet
Development

Successfully merging this pull request may close these issues.

2 participants