Skip to content

[fix] [tx] [branch-2.11] Transaction buffer recover blocked by readNext #18969

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.

Sign up for GitHub

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

Merged
congbobo184 merged 2 commits into from
Dec 19, 2022
Merged

[fix] [tx] [branch-2.11] Transaction buffer recover blocked by readNext #18969

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4d29bf3
[fix] [tx] [branch-2.11] Transaction buffer recover blocked by readNext
poorbarcode Dec 17, 2022
bb0359a
address comment: close reader
poorbarcode Dec 18, 2022
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
24 changes: 20 additions & 4 deletions ...rc/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicLong;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
Expand All @@ -52,13 +53,15 @@
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.transaction.TxnID;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.common.api.proto.MessageMetadata;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.TransactionBufferStats;
import org.apache.pulsar.common.policies.data.TransactionInBufferStats;
import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.protocol.Markers;
import org.apache.pulsar.common.util.Codec;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.RecoverTimeRecord;
import org.jctools.queues.MessagePassingQueue;
import org.jctools.queues.SpscArrayQueue;
Expand Down Expand Up @@ -615,6 +618,11 @@ private TopicTransactionBufferRecover(TopicTransactionBufferRecoverCallBack call
this.takeSnapshotWriter = takeSnapshotWriter;
}

private long getSystemClientOperationTimeoutMs() throws Exception {
PulsarClientImpl pulsarClient = (PulsarClientImpl) topic.getBrokerService().getPulsar().getClient();
return pulsarClient.getConfiguration().getOperationTimeoutMs();
}

@SneakyThrows
@Override
public void run() {
Expand All @@ -629,7 +637,8 @@ public void run() {
try {
boolean hasSnapshot = false;
while (reader.hasMoreEvents()) {
Message<TransactionBufferSnapshot> message = reader.readNext();
Message<TransactionBufferSnapshot> message = reader.readNextAsync()
.get(getSystemClientOperationTimeoutMs(), TimeUnit.MILLISECONDS);
if (topic.getName().equals(message.getKey())) {
TransactionBufferSnapshot transactionBufferSnapshot = message.getValue();
if (transactionBufferSnapshot != null) {
Expand All @@ -642,18 +651,25 @@ public void run() {
}
}
if (!hasSnapshot) {
closeReader(reader);
callBack.noNeedToRecover();
return;
}
} catch (TimeoutException ex) {
Throwable t = FutureUtil.unwrapCompletionException(ex);
String errorMessage = String.format("[%s] Transaction buffer recover fail by read "
+ "transactionBufferSnapshot timeout!", topic.getName());
log.error(errorMessage, t);
callBack.recoverExceptionally(
new BrokerServiceException.ServiceUnitNotReadyException(errorMessage, t));

@congbobo184 congbobo184 Dec 18, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should close the reader right?

@poorbarcode poorbarcode Dec 18, 2022

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, Already fixed.

return;
} catch (Exception ex) {
log.error("[{}] Transaction buffer recover fail when read "
+ "transactionBufferSnapshot!", topic.getName(), ex);
callBack.recoverExceptionally(ex);
closeReader(reader);
return;
} finally {
closeReader(reader);
}
closeReader(reader);

ManagedCursor managedCursor;
try {
Expand Down
81 changes: 81 additions & 0 deletions ...src/test/java/org/apache/pulsar/broker/transaction/TopicTransactionBufferRecoverTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,9 +20,12 @@

import static org.apache.pulsar.common.naming.SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.when;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
Expand All @@ -33,9 +36,11 @@
import java.lang.reflect.Method;
import java.util.NavigableMap;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
Expand Down Expand Up @@ -88,6 +93,7 @@ public class TopicTransactionBufferRecoverTest extends TransactionTestBase {
private static final int NUM_PARTITIONS = 16;
@BeforeMethod
protected void setup() throws Exception {
conf.getProperties().setProperty("brokerClient_operationTimeoutMs", Integer.valueOf(10 * 1000).toString());
setUpBase(1, NUM_PARTITIONS, RECOVER_COMMIT, 0);
admin.topics().createNonPartitionedTopic(RECOVER_ABORT);
admin.topics().createNonPartitionedTopic(TAKE_SNAPSHOT);
Expand Down Expand Up @@ -224,6 +230,81 @@ private void recoverTest(String testTopic) throws Exception {

}

private void makeTBSnapshotReaderTimeoutIfFirstRead(TopicName topicName) throws Exception {
SystemTopicClient.Reader mockReader = mock(SystemTopicClient.Reader.class);
AtomicBoolean isFirstCallOfMethodHasMoreEvents = new AtomicBoolean();
AtomicBoolean isFirstCallOfMethodHasReadNext = new AtomicBoolean();
AtomicBoolean isFirstCallOfMethodHasReadNextAsync = new AtomicBoolean();

doAnswer(invocation -> {
if (isFirstCallOfMethodHasMoreEvents.compareAndSet(false,true)){
return true;
} else {
return false;
}
}).when(mockReader).hasMoreEvents();

doAnswer(invocation -> {
if (isFirstCallOfMethodHasReadNext.compareAndSet(false, true)){
// Just stuck the thread.
Thread.sleep(3600 * 1000);
}
return null;
}).when(mockReader).readNext();

doAnswer(invocation -> {
CompletableFuture<Message> future = new CompletableFuture<>();
new Thread(() -> {
if (isFirstCallOfMethodHasReadNextAsync.compareAndSet(false, true)){
// Just stuck the thread.
try {
Thread.sleep(3600 * 1000);
} catch (InterruptedException e) {
}
future.complete(null);
} else {
future.complete(null);
}
}).start();
return future;
}).when(mockReader).readNextAsync();

when(mockReader.closeAsync()).thenReturn(CompletableFuture.completedFuture(null));

for (PulsarService pulsarService : pulsarServiceList){
// Init prop: lastMessageIdInBroker.
final TransactionBufferSnapshotService tbSnapshotService =
pulsarService.getTransactionBufferSnapshotService();
TransactionBufferSnapshotService spyTbSnapshotService = spy(tbSnapshotService);
doAnswer(invocation -> CompletableFuture.completedFuture(mockReader))
.when(spyTbSnapshotService).createReader(topicName);
Field field =
PulsarService.class.getDeclaredField("transactionBufferSnapshotService");
field.setAccessible(true);
field.set(pulsarService, spyTbSnapshotService);
}
}

@Test(timeOut = 60 * 1000)
public void testTBRecoverCanRetryIfTimeoutRead() throws Exception {
String topicName = String.format("persistent://%s/%s", NAMESPACE1,
"tx_recover_" + UUID.randomUUID().toString().replaceAll("-", "_"));

// Make race condition of "getLastMessageId" and "compaction" to make recover can't complete.
makeTBSnapshotReaderTimeoutIfFirstRead(TopicName.get(topicName));
// Verify( Cmd-PRODUCER will wait for TB recover finished )
Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
.topic(topicName)
.sendTimeout(0, TimeUnit.SECONDS)
.enableBatching(false)
.batchingMaxMessages(2)
.create();

// cleanup.
producer.close();
admin.topics().delete(topicName, false);
}

@Test
private void testTakeSnapshot() throws IOException, ExecutionException, InterruptedException {

Expand Down