refactor(engine): thread-safe transient subscription state #13072
Conversation
This changes the existing transient state holder to be thread safe. Allows for future changes where we need concurrent read and write operations.
The previous class hierarchy was a bit confusing. With these changes here, the `DbMessageSubscriptionState` is responsible for all state changes. When necessary, it reads from and writes to the transient state directly. This also adjusts names and package paths to indicate that reading and writing transient state is not a mutable operation.
The previous class hierarchy was a bit confusing. With these changes here, the `DbProcessMessageSubscriptionState` is responsible for all state changes. When necessary, it reads from and writes to the transient state directly. This also adjusts names and package paths to indicate that reading and writing transient state is not a mutable operation.
oleschoenburg
force-pushed
the
os/fix-mutable-pending-msg-state
branch
from
9afa964
to
f8a7dc4
Compare
|
Thanks @oleschoenburg ❤️ I'll have a look at this tomorrow |
|
Sorry @oleschoenburg, I forgot about this one. It'll be my top priority tomorrow 🙇 |
There was a problem hiding this comment.
Thanks @oleschoenburg 👏
👍 Another amazing set of changes!
❌ I'm not confident about the thread safety of the TransientPendingSubscriptionState. I might misunderstand how the ConcurrentHashMap works, but please have another look.
...e/src/main/java/io/camunda/zeebe/engine/state/message/TransientPendingSubscriptionState.java
Show resolved
Hide resolved
| @@ -151,13 +155,20 @@ public void updateToCorrelatingState(final MessageSubscriptionRecord record) { | |||
|
|
|||
| updateCorrelatingFlag(subscription, true); | |||
|
|
|||
| transientState.add(record); | |||
| transientState.add( | |||
There was a problem hiding this comment.
🔧 Although the implementation of add and update are equivalent, semantically, it might make more sense to use update here.
| } | ||
|
|
||
| @Override | ||
| public void updateToOpeningState(final ProcessMessageSubscriptionRecord record) { | ||
| update(record, s -> s.setRecord(record).setOpening()); | ||
| transientState.add(record); | ||
| transientState.add( |
There was a problem hiding this comment.
🔧 Although the implementation of add and update are equivalent, semantically, it might make more sense to use update here.
| } | ||
|
|
||
| @Override | ||
| public void updateToClosingState(final ProcessMessageSubscriptionRecord record) { | ||
| update(record, s -> s.setRecord(record).setClosing()); | ||
| transientState.add(record); | ||
| transientState.add( |
There was a problem hiding this comment.
🔧 Although the implementation of add and update are equivalent, semantically, it might make more sense to use update here.
There was a problem hiding this comment.
@oleschoenburg You have convinced me here that this is indeed thread safe.
👍 LGTM (please consider my suggestion before merging)
Although the implementation is equivalent, using `update` instead of `add` is clearer.
|
bors merge |
|
Build succeeded: |
Closes #12798 by making
TransientSubscriptionCommandStatethread safe. This allows data sharing between scheduled tasks and processing.The real change is 839744f with two follow up commits that significantly simplify the class hierarchy but add lot of diff noise. The refactorings can be split off if necessary.