Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[fix][broker] PIP-379 Key_Shared implementation race condition causing out-of-order message delivery #23874

Merged
Merged
Changes from 1 commit
Commits
Show all changes
28 commits
Select commit Hold shift + click to select a range
67605e7
Add solution to PulsarMockBookKeeper for intercepting reads
lhotari Oct 21, 2024
e40461f
Improve quiet time implementation in receiveMessages
lhotari Jan 17, 2025
7e69af5
Add debug when added to replay
lhotari Jan 17, 2025
6a354df
Enable test logging at debug level, add more logging
lhotari Jan 20, 2025
7c1d3db
Cancel pending read
lhotari Jan 20, 2025
47f7583
Add debug log to skipping pending replay read
lhotari Jan 21, 2025
4dc843b
Add test
lhotari Jan 16, 2025
0512f38
Notify also when the consumer isn't closing
lhotari Jan 21, 2025
3913e64
Postpone removals after critical sections to prevent race conditions
lhotari Jan 21, 2025
fa60a7a
Add test and more docs for OutsideCriticalSectionsExecutor
lhotari Jan 21, 2025
ec898d6
Adjust test logging - disable debug logging for key_shared related di…
lhotari Jan 21, 2025
7bc7b92
Fix failing test
lhotari Jan 21, 2025
7bca0ad
Fix race condition in test
lhotari Jan 21, 2025
c8d14f3
Revert "Add solution to PulsarMockBookKeeper for intercepting reads"
lhotari Jan 22, 2025
799fab9
Revert "Improve quiet time implementation in receiveMessages"
lhotari Jan 22, 2025
d8361bf
Fix test after reverting read handle interceptor changes
lhotari Jan 22, 2025
eef83e9
Revert "Add test and more docs for OutsideCriticalSectionsExecutor"
lhotari Jan 22, 2025
31db5fa
Revert "Postpone removals after critical sections to prevent race con…
lhotari Jan 22, 2025
231b300
Fix test after reverting previous changes
lhotari Jan 22, 2025
e734cdf
Add solution by Yubiao to prevent race condition in unblocking while …
lhotari Jan 22, 2025
a5d2029
Add "already blocked hashes" solution to dispatching phase
lhotari Jan 22, 2025
450069a
Run test with most classic and PIP-379 implementation
lhotari Jan 22, 2025
06827df
Revert havePendingReplayRead related changes
lhotari Jan 23, 2025
98942d7
Validate that no exceptions were thrown in message handler
lhotari Jan 23, 2025
65401dc
Make failing configurable when out-of-order message is a duplicate
lhotari Jan 23, 2025
ddbe35b
Improve test logging
lhotari Jan 23, 2025
4b0f08b
Remove unused import
lhotari Jan 23, 2025
be9b80c
Revert unrelated DrainingHashesTracker changes
lhotari Jan 23, 2025
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Validate that no exceptions were thrown in message handler
lhotari committed Jan 23, 2025
commit 98942d7d12d33602999489017db511eb6d62c24c
Original file line number Diff line number Diff line change
@@ -21,6 +21,7 @@
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.apache.pulsar.broker.BrokerTestUtil.newUniqueName;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.SoftAssertions.assertSoftly;
import static org.testng.Assert.fail;
import java.time.Duration;
import java.util.ArrayList;
@@ -189,33 +190,42 @@ public void testMessageOrderInSingleConsumerReconnect(KeySharedImplementationTyp
AtomicBoolean c2MessagesShouldBeUnacked = new AtomicBoolean(true);
Set<String> keysForC2 = new HashSet<>();
AtomicLong lastMessageTimestamp = new AtomicLong(System.currentTimeMillis());
List<Throwable> exceptionsInHandler = Collections.synchronizedList(new ArrayList<>());

Map<String, Pair<Position, String>> keyPositions = new HashMap<>();
MessageListener<Integer> messageHandler = (consumer, msg) -> {
lastMessageTimestamp.set(System.currentTimeMillis());
synchronized (this) {
String key = msg.getKey();
if (c2MessagesShouldBeUnacked.get() && keysForC2.contains(key)) {
unackedMessages.add(Pair.of(consumer, msg));
return;
}
long delayMillis = ThreadLocalRandom.current().nextLong(25, 50);
CompletableFuture.delayedExecutor(delayMillis, TimeUnit.MILLISECONDS).execute(() ->
consumer.acknowledgeAsync(msg));
MessageIdAdv msgId = (MessageIdAdv) msg.getMessageId();
Position currentPosition = PositionFactory.create(msgId.getLedgerId(), msgId.getEntryId());
Pair<Position, String> prevPair = keyPositions.get(key);
if (prevPair != null && prevPair.getLeft().compareTo(currentPosition) > 0) {
boolean isDuplicate = !remainingMessageValues.contains(msg.getValue());
log.error("key: {} value: {} prev: {}/{} current: {}/{} duplicate: {}", key, msg.getValue(), prevPair.getLeft(),
prevPair.getRight(), currentPosition, consumer.getConsumerName(), isDuplicate);
fail("out of order");
}
keyPositions.put(key, Pair.of(currentPosition, consumer.getConsumerName()));
boolean removed = remainingMessageValues.remove(msg.getValue());
if (!removed) {
// duplicates are possible during reconnects, this is not an error
log.warn("Duplicate message: {} value: {}", msg.getMessageId(), msg.getValue());
try {
String key = msg.getKey();
if (c2MessagesShouldBeUnacked.get() && keysForC2.contains(key)) {
unackedMessages.add(Pair.of(consumer, msg));
return;
}
long delayMillis = ThreadLocalRandom.current().nextLong(25, 50);
CompletableFuture.delayedExecutor(delayMillis, TimeUnit.MILLISECONDS).execute(() ->
consumer.acknowledgeAsync(msg));
MessageIdAdv msgId = (MessageIdAdv) msg.getMessageId();
Position currentPosition = PositionFactory.create(msgId.getLedgerId(), msgId.getEntryId());
Pair<Position, String> prevPair = keyPositions.get(key);
if (prevPair != null && prevPair.getLeft().compareTo(currentPosition) > 0) {
boolean isDuplicate = !remainingMessageValues.contains(msg.getValue());
log.error("key: {} value: {} prev: {}/{} current: {}/{} duplicate: {}", key, msg.getValue(),
prevPair.getLeft(),
prevPair.getRight(), currentPosition, consumer.getConsumerName(), isDuplicate);
fail("out of order");
}
keyPositions.put(key, Pair.of(currentPosition, consumer.getConsumerName()));
boolean removed = remainingMessageValues.remove(msg.getValue());
if (!removed) {
// duplicates are possible during reconnects, this is not an error
log.warn("Duplicate message: {} value: {}", msg.getMessageId(), msg.getValue());
}
} catch (Throwable t) {
exceptionsInHandler.add(t);
if (!(t instanceof AssertionError)) {
log.error("Error in message handler", t);
}
}
}
};
@@ -320,7 +330,10 @@ public void testMessageOrderInSingleConsumerReconnect(KeySharedImplementationTyp
});

try {
assertThat(remainingMessageValues).isEmpty();
assertSoftly(softly -> {
softly.assertThat(remainingMessageValues).as("remainingMessageValues").isEmpty();
softly.assertThat(exceptionsInHandler).as("exceptionsInHandler").isEmpty();
});
} finally {
logTopicStats(topic);
}