-
Notifications
You must be signed in to change notification settings - Fork 565
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
8798: Add API to probe the logstream batch writer if more bytes can be written without writing them r=npepinpe a=npepinpe ## Description This PR adds a new API method, `LogStreamBatchWriter#canWriteAdditionalEvent(int)`. This allows users of the writer to probe if adding the given amount of bytes to the batch would cause it to become un-writable, without actually having to write anything to the batch, or even modify their DTO (e.g. the `TypedRecord<?>` in the engine). To avoid having dispatcher details leak into the implementation, an analogous method is added to the dispatcher, `Dispatcher#canClaimFragmentBatch(int, int)`, which will compare the given size, framed and aligned, with the max fragment length. This is the main building block to eventually solve #5525, and enable other use cases (e.g. multi-instance creation) which deal with large batches until we have a more permanent solution (e.g. chunking follow up batches). NOTE: the tests added in the dispatcher are not very good, but I couldn't come up with something else that wouldn't be too coupled to the implementation (i.e. essentially reusing `LogBufferAppender`). I would like some ideas/suggestions. NOTE: this PR comes out of the larger one, #8491. You can check that one out to see how the new API would be used, e.g. in the `JobBatchCollector`. As such, this is marked for backporting, since we'll backport the complete fix for #5525. ## Related issues related to #5525 Co-authored-by: Nicolas Pepin-Perreault <nicolas.pepin-perreault@camunda.com>
- Loading branch information
Showing
6 changed files
with
175 additions
and
17 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
77 changes: 77 additions & 0 deletions
77
...eams/src/test/java/io/camunda/zeebe/logstreams/impl/log/LogStreamBatchWriterImplTest.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,77 @@ | ||
| /* | ||
| * Copyright Camunda Services GmbH and/or licensed to Camunda Services GmbH under | ||
| * one or more contributor license agreements. See the NOTICE file distributed | ||
| * with this work for additional information regarding copyright ownership. | ||
| * Licensed under the Zeebe Community License 1.1. You may not use this file | ||
| * except in compliance with the Zeebe Community License 1.1. | ||
| */ | ||
| package io.camunda.zeebe.logstreams.impl.log; | ||
|
|
||
| import static org.assertj.core.api.Assertions.assertThat; | ||
| import static org.mockito.ArgumentMatchers.any; | ||
| import static org.mockito.ArgumentMatchers.anyInt; | ||
| import static org.mockito.ArgumentMatchers.eq; | ||
| import static org.mockito.Mockito.mock; | ||
| import static org.mockito.Mockito.times; | ||
| import static org.mockito.Mockito.verify; | ||
| import static org.mockito.Mockito.verifyNoMoreInteractions; | ||
| import static org.mockito.Mockito.when; | ||
|
|
||
| import io.camunda.zeebe.dispatcher.ClaimedFragmentBatch; | ||
| import io.camunda.zeebe.dispatcher.Dispatcher; | ||
| import io.camunda.zeebe.util.buffer.BufferUtil; | ||
| import org.agrona.DirectBuffer; | ||
| import org.agrona.ExpandableArrayBuffer; | ||
| import org.agrona.concurrent.UnsafeBuffer; | ||
| import org.junit.jupiter.api.Test; | ||
| import org.mockito.invocation.InvocationOnMock; | ||
|
|
||
| final class LogStreamBatchWriterImplTest { | ||
| private final Dispatcher dispatcher = mock(Dispatcher.class); | ||
| private final LogStreamBatchWriterImpl writer = new LogStreamBatchWriterImpl(1, dispatcher); | ||
|
|
||
| /** | ||
| * This test asserts that {@link LogStreamBatchWriterImpl#canWriteAdditionalEvent(int)} computes | ||
| * the correct batch length before passing it on to the dispatcher to check if it's acceptable. It | ||
| * also verifies that the same length is passed to the dispatcher when actually writing, ensuring | ||
| * both methods remain consistent with each other. | ||
| */ | ||
| @Test | ||
| void canWriteAdditionalEvent() { | ||
| // given | ||
| final DirectBuffer value = BufferUtil.wrapString("foo"); | ||
| final DirectBuffer metadata = BufferUtil.wrapString("bar"); | ||
| final var expectedFragmentCount = 2; | ||
| final var expectedFramingLength = | ||
| expectedFragmentCount * LogEntryDescriptor.HEADER_BLOCK_LENGTH; | ||
| final var expectedEventsLength = | ||
| expectedFragmentCount * (value.capacity() + metadata.capacity()); | ||
| final var expectedBatchLength = expectedEventsLength + expectedFramingLength; | ||
|
|
||
| // when | ||
| when(dispatcher.canClaimFragmentBatch(anyInt(), anyInt())).thenReturn(false); | ||
| when(dispatcher.canClaimFragmentBatch(expectedFragmentCount, expectedBatchLength)) | ||
| .thenReturn(true); | ||
| when(dispatcher.claimFragmentBatch(any(), anyInt(), anyInt())) | ||
| .then(this::mockClaimFragmentBatch); | ||
| writer.event().value(value).metadata(metadata).done(); | ||
| final boolean canWriteAdditionalEvent = | ||
| writer.canWriteAdditionalEvent(value.capacity() + metadata.capacity()); | ||
| writer.event().value(value).metadata(metadata).done().tryWrite(); | ||
|
|
||
| // then | ||
| verify(dispatcher, times(1)).canClaimFragmentBatch(expectedFragmentCount, expectedBatchLength); | ||
| verify(dispatcher, times(1)) | ||
| .claimFragmentBatch( | ||
| any(ClaimedFragmentBatch.class), eq(expectedFragmentCount), eq(expectedBatchLength)); | ||
| assertThat(canWriteAdditionalEvent).isTrue(); | ||
| verifyNoMoreInteractions(dispatcher); | ||
| } | ||
|
|
||
| private long mockClaimFragmentBatch(final InvocationOnMock i) { | ||
| final var batch = i.getArgument(0, ClaimedFragmentBatch.class); | ||
| final var writeBuffer = new UnsafeBuffer(new ExpandableArrayBuffer(1024)); | ||
| batch.wrap(writeBuffer, 1, 0, writeBuffer.capacity(), () -> {}); | ||
| return 1L; | ||
| } | ||
| } |