-
Notifications
You must be signed in to change notification settings - Fork 556
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
9840: `TypedStreamWriter` that writes to buffer r=pihme a=pihme ## Description - Implements `TypedStreamWriter` that writes to buffer - Implements processing result builder to use the buffered stream writer - does not add tests. I started it, but it would take too much time. Hope I can revisit it after my holidays. For now there is just a follow up issue: #9838 ## Related issues closes #9780 9844: Shape legacy code into new interfaces good bits part 2 r=pihme a=pihme ## Description - Move logic to select processor into engine - Move processing logic into engine - Use `ProcessingResult` to return response and execute side effects - Move error handling logic into engine - Fix tests ## Review Hints - most of the commits have some failing tests - this was unavoidable, because substeps in the modification were invalid/incomplete - In the end, all tests are now passing again. This was quite satisfying acutally - adding more changes and seeing the test failure count go down again - Overall I would say we are through the valley of pain. Was quite volatile recently, but I am confident we can bring some calmness back to development - This achieves that most of the engine abstraction classes are in play; and the division of labor is largely as intended. - Still lots of stuff to do on either side, though ## Related issues related to #9725 Co-authored-by: pihme <pihme@users.noreply.github.com>
- Loading branch information
Showing
19 changed files
with
512 additions
and
125 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
33 changes: 33 additions & 0 deletions
33
engine/src/main/java/io/camunda/zeebe/engine/api/EmptyProcessingResult.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,33 @@ | ||
| /* | ||
| * 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.engine.api; | ||
|
|
||
| import io.camunda.zeebe.engine.processing.streamprocessor.writers.CommandResponseWriter; | ||
| import io.camunda.zeebe.logstreams.log.LogStreamBatchWriter; | ||
|
|
||
| public final class EmptyProcessingResult implements ProcessingResult { | ||
|
|
||
| public static final ProcessingResult INSTANCE = new EmptyProcessingResult(); | ||
|
|
||
| private EmptyProcessingResult() {} | ||
|
|
||
| @Override | ||
| public long writeRecordsToStream(final LogStreamBatchWriter logStreamBatchWriter) { | ||
| return 0; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean writeResponse(final CommandResponseWriter commandResponseWriter) { | ||
| return true; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean executePostCommitTasks() { | ||
| return true; | ||
| } | ||
| } |
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
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
113 changes: 113 additions & 0 deletions
113
engine/src/main/java/io/camunda/zeebe/streamprocessor/BufferedProcessingResultBuilder.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,113 @@ | ||
| /* | ||
| * 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.streamprocessor; | ||
|
|
||
| import static io.camunda.zeebe.engine.processing.streamprocessor.TypedEventRegistry.EVENT_REGISTRY; | ||
|
|
||
| import io.camunda.zeebe.engine.api.PostCommitTask; | ||
| import io.camunda.zeebe.engine.api.ProcessingResult; | ||
| import io.camunda.zeebe.engine.api.ProcessingResultBuilder; | ||
| import io.camunda.zeebe.msgpack.UnpackedObject; | ||
| import io.camunda.zeebe.protocol.record.RecordType; | ||
| import io.camunda.zeebe.protocol.record.RecordValue; | ||
| import io.camunda.zeebe.protocol.record.RejectionType; | ||
| import io.camunda.zeebe.protocol.record.ValueType; | ||
| import io.camunda.zeebe.protocol.record.intent.Intent; | ||
| import io.camunda.zeebe.util.buffer.BufferWriter; | ||
| import java.util.ArrayList; | ||
| import java.util.HashMap; | ||
| import java.util.List; | ||
| import java.util.Map; | ||
| import java.util.function.BinaryOperator; | ||
|
|
||
| /** Implementation of {@code ProcessingResultBuilder} that writes all data into a buffer */ | ||
| final class BufferedProcessingResultBuilder implements ProcessingResultBuilder { | ||
|
|
||
| private final Map<Class<? extends UnpackedObject>, ValueType> typeRegistry; | ||
|
|
||
| private final BufferedStreamWriter bufferedStreamWriter; | ||
| private final List<PostCommitTask> postCommitTasks = new ArrayList<>(); | ||
| private final int sourceIndex; | ||
|
|
||
| BufferedProcessingResultBuilder( | ||
| final BinaryOperator<Integer> capacityCalculator, final int sourceIndex) { | ||
| bufferedStreamWriter = new BufferedStreamWriter(capacityCalculator); | ||
| this.sourceIndex = sourceIndex; | ||
|
|
||
| typeRegistry = new HashMap<>(); | ||
| EVENT_REGISTRY.forEach((e, c) -> typeRegistry.put(c, e)); | ||
| } | ||
|
|
||
| @Override | ||
| public ProcessingResultBuilder appendRecord( | ||
| final long key, | ||
| final RecordType type, | ||
| final Intent intent, | ||
| final RejectionType rejectionType, | ||
| final String rejectionReason, | ||
| final RecordValue value) { | ||
|
|
||
| final ValueType valueType = initValueType(value); | ||
| final var valueWriter = initValueWriter(value); | ||
|
|
||
| bufferedStreamWriter.appendRecord( | ||
| key, sourceIndex, type, intent, rejectionType, rejectionReason, valueType, valueWriter); | ||
| return this; | ||
| } | ||
|
|
||
| @Override | ||
| public ProcessingResultBuilder withResponse( | ||
| final long eventKey, | ||
| final Intent eventState, | ||
| final UnpackedObject eventValue, | ||
| final ValueType valueType, | ||
| final long requestId, | ||
| final int requestStreamId) { | ||
| throw new RuntimeException("Not yet implemented"); | ||
| } | ||
|
|
||
| @Override | ||
| public ProcessingResultBuilder appendPostCommitTask(final PostCommitTask r) { | ||
| postCommitTasks.add(r); | ||
| return this; | ||
| } | ||
|
|
||
| @Override | ||
| public ProcessingResultBuilder reset() { | ||
| bufferedStreamWriter.reset(); | ||
| postCommitTasks.clear(); | ||
| return this; | ||
| } | ||
|
|
||
| @Override | ||
| public ProcessingResult build() { | ||
| throw new RuntimeException("Not yet implemented"); | ||
| } | ||
|
|
||
| private ValueType initValueType(final RecordValue value) { | ||
| final ValueType valueType = typeRegistry.get(value.getClass()); | ||
| if (valueType == null) { | ||
| // usually happens when the record is not registered at the TypedStreamEnvironment | ||
| throw new IllegalArgumentException( | ||
| "Missing value type mapping for record: " + value.getClass()); | ||
| } | ||
| return valueType; | ||
| } | ||
|
|
||
| private BufferWriter initValueWriter(final RecordValue value) { | ||
| // TODO evaluate whether the interface should be changed to UnifiedRecordValue or <T extends | ||
| // RecordValue & BufferWriter> BufferWriter initValueWriter(final T value) {} | ||
| // validation | ||
| if (!(value instanceof BufferWriter)) { | ||
| throw new IllegalArgumentException( | ||
| String.format("The record value %s is not a BufferWriter", value)); | ||
| } | ||
|
|
||
| return (BufferWriter) value; | ||
| } | ||
| } |
Oops, something went wrong.