Skip to content

[fix] [tx] Transaction buffer recover blocked by readNext #18833

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
codelipenghui merged 8 commits into from
Dec 17, 2022
Merged

[fix] [tx] Transaction buffer recover blocked by readNext #18833

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
fc32b1f
[improve] [test] Add test testTrimLedgerWillKeepsAtLeastOneLedgerWith…
poorbarcode Dec 6, 2022
ac8c548
[fix] [tx] Transaction buffer recover can not completed
poorbarcode Dec 9, 2022
0dd2818
hanle npe more precision
poorbarcode Dec 9, 2022
cdff5e1
avoid unnecessary changes
poorbarcode Dec 10, 2022
73f90ef
Simplified logic
poorbarcode Dec 12, 2022
1718782
fix test
poorbarcode Dec 13, 2022
51d1f64
address comments
poorbarcode Dec 14, 2022
01d5cec
make test simpler
poorbarcode Dec 16, 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
25 changes: 20 additions & 5 deletions ...g/apache/pulsar/broker/transaction/buffer/impl/SingleSnapshotAbortedTxnProcessorImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,18 +21,22 @@
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.commons.collections4.map.LinkedMap;
import org.apache.pulsar.broker.service.BrokerServiceException;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.systopic.SystemTopicClient;
import org.apache.pulsar.broker.transaction.buffer.AbortedTxnProcessor;
import org.apache.pulsar.broker.transaction.buffer.metadata.AbortTxnMetadata;
import org.apache.pulsar.broker.transaction.buffer.metadata.TransactionBufferSnapshot;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.transaction.TxnID;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.util.FutureUtil;

Expand Down Expand Up @@ -78,16 +82,21 @@ public boolean checkAbortedTransaction(TxnID txnID, Position readPosition) {
return aborts.containsKey(txnID);
}

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

@Override
public CompletableFuture<PositionImpl> recoverFromSnapshot() {
return topic.getBrokerService().getPulsar().getTransactionBufferSnapshotServiceFactory()
.getTxnBufferSnapshotService()
.createReader(TopicName.get(topic.getName())).thenComposeAsync(reader -> {
PositionImpl startReadCursorPosition = null;
try {
PositionImpl startReadCursorPosition = null;
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 @@ -98,15 +107,21 @@ public CompletableFuture<PositionImpl> recoverFromSnapshot() {
}
}
}
closeReader(reader);
return CompletableFuture.completedFuture(startReadCursorPosition);
} 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);
return FutureUtil.failedFuture(
new BrokerServiceException.ServiceUnitNotReadyException(errorMessage, t));
} catch (Exception ex) {
log.error("[{}] Transaction buffer recover fail when read "
+ "transactionBufferSnapshot!", topic.getName(), ex);
closeReader(reader);
return FutureUtil.failedFuture(ex);
} finally {
closeReader(reader);
}

}, topic.getBrokerService().getPulsar().getTransactionExecutorProvider()
.getExecutor(this));
}
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 @@ -35,8 +38,10 @@
import java.util.List;
import java.util.NavigableMap;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.mledger.AsyncCallbacks;
Expand Down Expand Up @@ -104,6 +109,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 @@ -248,6 +254,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 SystemTopicTxnBufferSnapshotService tbSnapshotService =
pulsarService.getTransactionBufferSnapshotServiceFactory().getTxnBufferSnapshotService();
SystemTopicTxnBufferSnapshotService spyTbSnapshotService = spy(tbSnapshotService);
doAnswer(invocation -> CompletableFuture.completedFuture(mockReader))
.when(spyTbSnapshotService).createReader(topicName);
Field field =
TransactionBufferSnapshotServiceFactory.class.getDeclaredField("txnBufferSnapshotService");
field.setAccessible(true);
field.set(pulsarService.getTransactionBufferSnapshotServiceFactory(), 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);
}

private void testTakeSnapshot() throws Exception {
@Cleanup
Producer<String> producer = pulsarClient
Expand Down