You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by gh...@apache.org on 2023/08/10 21:02:48 UTC
[kafka] 02/02: KAFKA-15202: Fix MM2 offset translation when syncs are variably spaced (#14156)
This is an automated email from the ASF dual-hosted git repository.
gharris pushed a commit to branch 3.5
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 34a30fff5769bae55f978631e7a44f59631ee8a6
Author: Greg Harris <gr...@aiven.io>
AuthorDate: Thu Aug 10 13:25:50 2023 -0700
KAFKA-15202: Fix MM2 offset translation when syncs are variably spaced (#14156)
Reviewers: Chris Egerton <ch...@aiven.io>
---
.../kafka/connect/mirror/OffsetSyncStore.java | 61 +++++++++-----
.../kafka/connect/mirror/OffsetSyncStoreTest.java | 95 +++++++++++++++++-----
2 files changed, 116 insertions(+), 40 deletions(-)
diff --git a/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/OffsetSyncStore.java b/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/OffsetSyncStore.java
index 9222d314a43..63a91a11b45 100644
--- a/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/OffsetSyncStore.java
+++ b/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/OffsetSyncStore.java
@@ -189,7 +189,7 @@ class OffsetSyncStore implements AutoCloseable {
// Make a copy of the array before mutating it, so that readers do not see inconsistent data
// TODO: consider batching updates so that this copy can be performed less often for high-volume sync topics.
OffsetSync[] mutableSyncs = Arrays.copyOf(syncs, SYNCS_PER_PARTITION);
- updateSyncArray(mutableSyncs, offsetSync);
+ updateSyncArray(mutableSyncs, syncs, offsetSync);
if (log.isTraceEnabled()) {
log.trace("New sync {} applied, new state is {}", offsetSync, offsetArrayToString(mutableSyncs));
}
@@ -227,7 +227,7 @@ class OffsetSyncStore implements AutoCloseable {
}
}
- private void updateSyncArray(OffsetSync[] syncs, OffsetSync offsetSync) {
+ private void updateSyncArray(OffsetSync[] syncs, OffsetSync[] original, OffsetSync offsetSync) {
long upstreamOffset = offsetSync.upstreamOffset();
// While reading to the end of the topic, ensure that our earliest sync is later than
// any earlier sync that could have been used for translation, to preserve monotonicity
@@ -237,29 +237,49 @@ class OffsetSyncStore implements AutoCloseable {
return;
}
OffsetSync replacement = offsetSync;
- // The most-recently-discarded offset sync
- // We track this since it may still be eligible for use in the syncs array at a later index
- OffsetSync oldValue = syncs[0];
+ // Index into the original syncs array to the replacement we'll consider next.
+ int replacementIndex = 0;
// Invariant A is always violated once a new sync appears.
// Repair Invariant A: the latest sync must always be updated
syncs[0] = replacement;
for (int current = 1; current < SYNCS_PER_PARTITION; current++) {
int previous = current - 1;
- // We can potentially use oldValue instead of replacement, allowing us to keep more distinct values stored
- // If oldValue is not recent, it should be expired from the store
- boolean isRecent = invariantB(syncs[previous], oldValue, previous, current);
- // Ensure that this value is sufficiently separated from the previous value
- // We prefer to keep more recent syncs of similar precision (i.e. the value in replacement)
- boolean separatedFromPrevious = invariantC(syncs[previous], oldValue, previous);
- // Ensure that this value is sufficiently separated from the next value
- // We prefer to keep existing syncs of lower precision (i.e. the value in syncs[next])
- int next = current + 1;
- boolean separatedFromNext = next >= SYNCS_PER_PARTITION || invariantC(oldValue, syncs[next], current);
- // If this condition is false, oldValue will be expired from the store and lost forever.
- if (isRecent && separatedFromPrevious && separatedFromNext) {
- replacement = oldValue;
- }
+ // Try to choose a value from the old array as the replacement
+ // This allows us to keep more distinct values stored overall, improving translation.
+ boolean skipOldValue;
+ do {
+ OffsetSync oldValue = original[replacementIndex];
+ // If oldValue is not recent enough, then it is not valid to use at the current index.
+ // It may still be valid when used in a later index where values are allowed to be older.
+ boolean isRecent = invariantB(syncs[previous], oldValue, previous, current);
+ // Ensure that this value is sufficiently separated from the previous value
+ // We prefer to keep more recent syncs of similar precision (i.e. the value in replacement)
+ // If this value is too close to the previous value, it will never be valid in a later position.
+ boolean separatedFromPrevious = invariantC(syncs[previous], oldValue, previous);
+ // Ensure that this value is sufficiently separated from the next value
+ // We prefer to keep existing syncs of lower precision (i.e. the value in syncs[next])
+ int next = current + 1;
+ boolean separatedFromNext = next >= SYNCS_PER_PARTITION || invariantC(oldValue, syncs[next], current);
+ // If the next value in the old array is a duplicate of the current one, then they are equivalent
+ // This value will not need to be considered again
+ int nextReplacement = replacementIndex + 1;
+ boolean duplicate = nextReplacement < SYNCS_PER_PARTITION && oldValue == original[nextReplacement];
+
+ // Promoting the oldValue to the replacement only happens if it satisfies all invariants.
+ boolean promoteOldValueToReplacement = isRecent && separatedFromPrevious && separatedFromNext;
+ if (promoteOldValueToReplacement) {
+ replacement = oldValue;
+ }
+ // The index should be skipped without promoting if we know that it will not be used at a later index
+ // based only on the observed part of the array so far.
+ skipOldValue = duplicate || !separatedFromPrevious;
+ if (promoteOldValueToReplacement || skipOldValue) {
+ replacementIndex++;
+ }
+ // We may need to skip past multiple indices, so keep looping until we're done skipping forward.
+ // The index must not get ahead of the current index, as we only promote from low index to high index.
+ } while (replacementIndex < current && skipOldValue);
// The replacement variable always contains a value which satisfies the invariants for this index.
// This replacement may or may not be used, since the invariants could already be satisfied,
@@ -273,8 +293,7 @@ class OffsetSyncStore implements AutoCloseable {
break;
} else {
// Invariant B violated for syncs[current]: sync is now too old and must be updated
- // Repair Invariant B: swap in replacement, and save the old value for the next iteration
- oldValue = syncs[current];
+ // Repair Invariant B: swap in replacement
syncs[current] = replacement;
assert invariantB(syncs[previous], syncs[current], previous, current);
diff --git a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/OffsetSyncStoreTest.java b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/OffsetSyncStoreTest.java
index b2fa9fa4da3..05c13f38b5b 100644
--- a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/OffsetSyncStoreTest.java
+++ b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/OffsetSyncStoreTest.java
@@ -22,6 +22,9 @@ import org.apache.kafka.common.TopicPartition;
import org.junit.jupiter.api.Test;
import java.util.OptionalLong;
+import java.util.PrimitiveIterator;
+import java.util.Random;
+import java.util.stream.LongStream;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -153,31 +156,85 @@ public class OffsetSyncStoreTest {
}
@Test
- public void testKeepMostDistinctSyncs() {
+ public void testConsistentlySpacedSyncs() {
// Under normal operation, the incoming syncs will be regularly spaced and the store should keep a set of syncs
// which provide the best translation accuracy (expires as few syncs as possible)
- // Each new sync should be added to the cache and expire at most one other sync from the cache
- long iterations = 10000;
+ long iterations = 100;
long maxStep = Long.MAX_VALUE / iterations;
// Test a variety of steps (corresponding to the offset.lag.max configuration)
for (long step = 1; step < maxStep; step = (step * 2) + 1) {
for (long firstOffset = 0; firstOffset < 30; firstOffset++) {
- try (FakeOffsetSyncStore store = new FakeOffsetSyncStore()) {
- int lastCount = 1;
- store.start();
- for (long offset = firstOffset; offset <= iterations; offset += step) {
- store.sync(tp, offset, offset);
- // Invariant A: the latest sync is present
- assertEquals(offset, store.syncFor(tp, 0).upstreamOffset());
- // Invariant D: the earliest sync is present
- assertEquals(firstOffset, store.syncFor(tp, 63).upstreamOffset());
- int count = countDistinctStoredSyncs(store, tp);
- int diff = count - lastCount;
- assertTrue(diff >= 0,
- "Store expired too many syncs: " + diff + " after receiving offset " + offset);
- lastCount = count;
- }
- }
+ long finalStep = step;
+ // Generate a stream of consistently spaced syncs
+ // Each new sync should be added to the cache and expire at most one other sync from the cache
+ assertSyncSpacingHasBoundedExpirations(firstOffset, LongStream.generate(() -> finalStep).limit(iterations), 1);
+ }
+ }
+ }
+
+ @Test
+ public void testRandomlySpacedSyncs() {
+ Random random = new Random(0L); // arbitrary but deterministic seed
+ int iterationBits = 10;
+ long iterations = 1 << iterationBits;
+ for (int n = 1; n < Long.SIZE - iterationBits; n++) {
+ // A stream with at most n bits of difference between the largest and smallest steps
+ // will expire n + 2 syncs at once in the worst case, because the sync store is laid out exponentially.
+ long maximumDifference = 1L << n;
+ int maximumExpirations = n + 2;
+ assertSyncSpacingHasBoundedExpirations(0, random.longs(iterations, 0L, maximumDifference), maximumExpirations);
+ // This holds true even if there is a larger minimum step size, such as caused by offsetLagMax
+ long offsetLagMax = 1L << 16;
+ assertSyncSpacingHasBoundedExpirations(0, random.longs(iterations, offsetLagMax, offsetLagMax + maximumDifference), maximumExpirations);
+ }
+ }
+
+ @Test
+ public void testDroppedSyncsSpacing() {
+ Random random = new Random(0L); // arbitrary but deterministic seed
+ long iterations = 10000;
+ long offsetLagMax = 100;
+ // Half of the gaps will be offsetLagMax, and half will be double that, as if one intervening sync was dropped.
+ LongStream stream = random.doubles()
+ .mapToLong(d -> (d < 0.5 ? 2 : 1) * offsetLagMax)
+ .limit(iterations);
+ // This will cause up to 2 syncs to be discarded, because a sequence of two adjacent syncs followed by a
+ // dropped sync will set up the following situation
+ // before [d....d,c,b,a....]
+ // after [e......e,d,a....]
+ // and syncs b and c are discarded to make room for e and the demoted sync d.
+ assertSyncSpacingHasBoundedExpirations(0, stream, 2);
+ }
+
+ /**
+ * Simulate an OffsetSyncStore receiving a sequence of offset syncs as defined by their start offset and gaps.
+ * After processing each simulated sync, assert that the store has not expired more unique syncs than the bound.
+ * @param firstOffset First offset to give to the sync store after starting
+ * @param steps A finite stream of gaps between syncs with some known distribution
+ * @param maximumExpirations The maximum number of distinct syncs allowed to be expired after a single update.
+ */
+ private void assertSyncSpacingHasBoundedExpirations(long firstOffset, LongStream steps, int maximumExpirations) {
+ try (FakeOffsetSyncStore store = new FakeOffsetSyncStore()) {
+ store.start();
+ store.sync(tp, firstOffset, firstOffset);
+ PrimitiveIterator.OfLong iterator = steps.iterator();
+ long offset = firstOffset;
+ int lastCount = 1;
+ while (iterator.hasNext()) {
+ offset += iterator.nextLong();
+ assertTrue(offset >= 0, "Test is invalid, offset overflowed");
+ store.sync(tp, offset, offset);
+ // Invariant A: the latest sync is present
+ assertEquals(offset, store.syncFor(tp, 0).upstreamOffset());
+ // Invariant D: the earliest sync is present
+ assertEquals(firstOffset, store.syncFor(tp, 63).upstreamOffset());
+ int count = countDistinctStoredSyncs(store, tp);
+ // We are adding one sync, so if the count didn't change, then exactly one sync expired.
+ int expiredSyncs = lastCount - count + 1;
+ assertTrue(expiredSyncs <= maximumExpirations,
+ "Store expired too many syncs: " + expiredSyncs + " > " + maximumExpirations
+ + " after receiving offset " + offset);
+ lastCount = count;
}
}
}