congbobo184 commented on a change in pull request #9490:
URL: https://github.com/apache/pulsar/pull/9490#discussion_r575820695
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TopicTransactionBuffer.java
##########
@@ -207,6 +371,190 @@ public void
syncMaxReadPositionForNormalPublish(PositionImpl position) {
@Override
public PositionImpl getMaxReadPosition() {
- return this.maxReadPosition;
+ if (checkIfReady()) {
+ return this.maxReadPosition;
+ } else {
+ return PositionImpl.earliest;
+ }
+ }
+
+ @Override
+ public void run(Timeout timeout) {
+ if (checkIfReady()) {
+ takeSnapshotByTimeout();
+ this.timer.newTimeout(this, takeSnapshotIntervalTime,
TimeUnit.MILLISECONDS);
+ }
+ }
+
+ static class TopicTransactionBufferRecover implements Runnable {
+
+ private final PersistentTopic topic;
+
+ private final TopicTransactionBufferRecoverCallBack callBack;
+
+ private Position startReadCursorPosition = PositionImpl.earliest;
+
+ private final SpscArrayQueue<Entry> entryQueue;
+
+ private final AtomicLong exceptionNumber = new AtomicLong();
+
+ // TODO: MAX_EXCEPTION_NUMBER can config
+ private static final int MAX_EXCEPTION_NUMBER = 500;
+
+ public static final String SUBSCRIPTION_NAME =
"transaction-buffer-sub";
+
+ private final TopicTransactionBuffer topicTransactionBuffer;
+
+ private
TopicTransactionBufferRecover(TopicTransactionBufferRecoverCallBack callBack,
PersistentTopic topic,
+ TopicTransactionBuffer
transactionBuffer) {
+ this.topic = topic;
+ this.callBack = callBack;
+ this.entryQueue = new SpscArrayQueue<>(2000);
+ this.topicTransactionBuffer = transactionBuffer;
+ }
+
+ @SneakyThrows
+ @Override
+ public void run() {
+ this.topicTransactionBuffer.changeToInitializingState();
+
topic.getBrokerService().getPulsar().getTransactionBufferSnapshotService()
+
.createReader(TopicName.get(topic.getName())).thenAcceptAsync(reader -> {
+ try {
+ while (reader.hasMoreEvents()) {
+ Message<TransactionBufferSnapshot> message =
reader.readNext();
+ TransactionBufferSnapshot transactionBufferSnapshot =
message.getValue();
+ if
(topic.getName().equals(transactionBufferSnapshot.getTopicName())) {
+ callBack.handleSnapshot(transactionBufferSnapshot);
+ this.startReadCursorPosition = PositionImpl.get(
+
transactionBufferSnapshot.getMaxReadPositionLedgerId(),
+
transactionBufferSnapshot.getMaxReadPositionEntryId());
+ }
+ }
+ } catch (PulsarClientException pulsarClientException) {
+ log.error("[{}]Transaction buffer recover fail when read "
+ + "transactionBufferSnapshot!", topic.getName(),
pulsarClientException);
+ reader.closeAsync().exceptionally(e -> {
+ log.error("[{}]Transaction buffer reader close
error!", topic.getName(), e);
+ return null;
+ });
+ return;
+ }
+ reader.closeAsync().exceptionally(e -> {
+ log.error("[{}]Transaction buffer reader close error!",
topic.getName(), e);
+ return null;
+ });
+
+ ManagedCursor managedCursor;
+ try {
+ managedCursor = topic.getManagedLedger()
+ .newNonDurableCursor(this.startReadCursorPosition,
SUBSCRIPTION_NAME);
+ } catch (ManagedLedgerException e) {
+ log.error("[{}]Transaction buffer recover fail when open
cursor!", topic.getName(), e);
+ return;
+ }
+ PositionImpl lastConfirmedEntry = (PositionImpl)
topic.getManagedLedger().getLastConfirmedEntry();
+ PositionImpl currentLoadPosition = (PositionImpl)
this.startReadCursorPosition;
+ FillEntryQueueCallback fillEntryQueueCallback = new
FillEntryQueueCallback(entryQueue, managedCursor,
+ TopicTransactionBufferRecover.this);
+ if (lastConfirmedEntry.getEntryId() != -1) {
+ while (lastConfirmedEntry.compareTo(currentLoadPosition) >
0) {
+ fillEntryQueueCallback.fillQueue();
+ Entry entry = entryQueue.poll();
+ if (entry != null) {
+ try {
+ currentLoadPosition =
PositionImpl.get(entry.getLedgerId(), entry.getEntryId());
+ callBack.handleTxnEntry(entry);
+ } finally {
+ entry.release();
+ }
+ } else {
+ if (exceptionNumber.get() > MAX_EXCEPTION_NUMBER) {
+ log.error("[{}]Transaction buffer recover fail
when "
Review comment:
this is poll method, when read operation not complete, the entry will be
null.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
