From 5fe1871bd9438bff0a68392a7b08166e4bcdaa2c Mon Sep 17 00:00:00 2001 From: Antonio Perez Dieppa Date: Sun, 15 Nov 2026 17:27:30 +0800 Subject: [PATCH] refactor: add DuplicateStreamSequenceException --- .../DuplicateStreamSequenceException.java | 87 ++++++++++++++++++ .../DuplicateStreamSequenceExceptionTest.java | 88 +++++++++++++++++++ 2 files changed, 175 insertions(+) create mode 100644 core/flamingock-core-commons/src/main/java/io/flamingock/internal/common/core/journal/DuplicateStreamSequenceException.java create mode 100644 core/flamingock-core-commons/src/test/java/io/flamingock/internal/common/core/journal/DuplicateStreamSequenceExceptionTest.java diff --git a/core/flamingock-core-commons/src/main/java/io/flamingock/internal/common/core/journal/DuplicateStreamSequenceException.java b/core/flamingock-core-commons/src/main/java/io/flamingock/internal/common/core/journal/DuplicateStreamSequenceException.java new file mode 100644 index 000000000..7f9c9c024 --- /dev/null +++ b/core/flamingock-core-commons/src/main/java/io/flamingock/internal/common/core/journal/DuplicateStreamSequenceException.java @@ -0,0 +1,87 @@ +/* + * Copyright 2026 Flamingock (https://www.flamingock.io) + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.flamingock.internal.common.core.journal; + +import io.flamingock.internal.common.core.error.FlamingockException; + +/** + * Signals that a journal event could not be appended because its + * {@code (streamId, streamSequence)} position was already taken. + *

+ * Every audit store already rejects a duplicate position — MongoDB through a unique index on + * {@code (streamId, streamSequence)} plus an insert that is never an upsert, SQL through + * {@code PRIMARY KEY (stream_id, stream_sequence)} and a plain {@code INSERT}, Couchbase because the + * document key is the position and it inserts rather than upserts, and DynamoDB because the pair + * is the table's primary key, guarded by a condition expression. What they do not share is how + * they say so: {@code DuplicateKeyException}, a {@code SQLException}, {@code DocumentExistsException}, or + * a {@code ConditionalCheckFailedException} buried inside a transaction cancellation. + *

+ * This is the common type they translate to, so a caller can tell "that position is taken" apart from + * "the write failed for some other reason" without knowing which backend it is talking to. That + * distinction is what makes a portable retry possible: retrying a genuine infrastructure failure would + * be wrong, and retrying nothing at all would leave a recoverable collision as a hard failure. + *

+ * Nothing throws this yet. Each store translates its own exception in its own change; see + * {@code docs/AUDIT_COMPACTION_CONTRIBUTOR_GUIDE.md} for how the journal work is split. + * + *

Why it carries the position

+ * A retry needs to know which position was rejected, and an operator reading a log needs both halves: + * {@code streamSequence} on its own identifies nothing, since every stream has its own sequence 1. + */ +public class DuplicateStreamSequenceException extends FlamingockException { + + private final String streamId; + private final long streamSequence; + + /** + * @param streamId the stream whose position was taken + * @param streamSequence the position within that stream + * @param cause the backend's own exception. Worth keeping rather than swallowing: without it, + * a position collision and a database rejecting the write for an unrelated + * reason are indistinguishable after the fact + */ + public DuplicateStreamSequenceException(String streamId, long streamSequence, Throwable cause) { + super(message(streamId, streamSequence), cause); + this.streamId = streamId; + this.streamSequence = streamSequence; + } + + /** + * For the case where a store detects the duplicate without an exception to wrap — for instance from an + * affected-row count, or a conditional write that reports failure as a value. + * + * @param streamId the stream whose position was taken + * @param streamSequence the position within that stream + */ + public DuplicateStreamSequenceException(String streamId, long streamSequence) { + super(message(streamId, streamSequence)); + this.streamId = streamId; + this.streamSequence = streamSequence; + } + + public String getStreamId() { + return streamId; + } + + public long getStreamSequence() { + return streamSequence; + } + + private static String message(String streamId, long streamSequence) { + return "Journal event position already taken [streamId=" + streamId + + " streamSequence=" + streamSequence + "]"; + } +} diff --git a/core/flamingock-core-commons/src/test/java/io/flamingock/internal/common/core/journal/DuplicateStreamSequenceExceptionTest.java b/core/flamingock-core-commons/src/test/java/io/flamingock/internal/common/core/journal/DuplicateStreamSequenceExceptionTest.java new file mode 100644 index 000000000..f33f03f0f --- /dev/null +++ b/core/flamingock-core-commons/src/test/java/io/flamingock/internal/common/core/journal/DuplicateStreamSequenceExceptionTest.java @@ -0,0 +1,88 @@ +/* + * Copyright 2026 Flamingock (https://www.flamingock.io) + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.flamingock.internal.common.core.journal; + +import io.flamingock.internal.common.core.error.FlamingockException; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * There is little behaviour in an exception class, so these cover only what something else depends on: + * the position the retry reads back, the cause a diagnosis needs, and the message an operator sees. + */ +class DuplicateStreamSequenceExceptionTest { + + @Test + @DisplayName("the message names both halves of the position") + void messageNamesTheWholePosition() { + DuplicateStreamSequenceException exception = + new DuplicateStreamSequenceException("stage-one", 42L, new IllegalStateException("taken")); + + String message = exception.getMessage(); + // Both halves, because a sequence on its own identifies nothing: every stream has a position 1. + // This message is all an operator gets from a log line. + assertTrue(message.contains("stage-one"), message); + assertTrue(message.contains("42"), message); + } + + @Test + @DisplayName("the backend's own exception is preserved as the cause") + void preservesTheBackendCause() { + IllegalStateException backendFailure = new IllegalStateException("E11000 duplicate key"); + + DuplicateStreamSequenceException exception = + new DuplicateStreamSequenceException("stage-one", 42L, backendFailure); + + // The one that matters. Swallowing it would make a position collision and a database rejecting + // the write for an unrelated reason look identical after the fact. + assertSame(backendFailure, exception.getCause()); + } + + @Test + @DisplayName("the position is readable, which is what a retry needs") + void exposesThePosition() { + DuplicateStreamSequenceException exception = + new DuplicateStreamSequenceException("stage-one", 42L, new IllegalStateException()); + + assertEquals("stage-one", exception.getStreamId()); + assertEquals(42L, exception.getStreamSequence()); + } + + @Test + @DisplayName("a store with no exception to wrap still reports the position") + void supportsDetectionWithoutABackendException() { + // For backends that report a rejected conditional write as a value rather than by throwing. + DuplicateStreamSequenceException exception = + new DuplicateStreamSequenceException("stage-one", 42L); + + assertNull(exception.getCause()); + assertEquals("stage-one", exception.getStreamId()); + assertEquals(42L, exception.getStreamSequence()); + } + + @Test + @DisplayName("it is a FlamingockException, so existing handlers see it") + void isAFlamingockException() { + // Relevant if it ever escapes un-retried: it surfaces as a Flamingock error rather than a bare + // runtime exception. + assertTrue(new DuplicateStreamSequenceException("stage-one", 1L) instanceof FlamingockException); + } +}