You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2022/04/06 12:05:43 UTC
[GitHub] [pulsar] mattisonchao commented on a diff in pull request #15015: [enh][transaction] Optimize to reuse transaction buffer snapshot writer
mattisonchao commented on code in PR #15015:
URL: https://github.com/apache/pulsar/pull/15015#discussion_r843714614
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBaseTxnBufferSnapshotService.java:
##########
@@ -18,69 +18,176 @@
*/
package org.apache.pulsar.broker.service;
+import io.netty.util.AbstractReferenceCounted;
+import io.netty.util.ReferenceCounted;
+import java.util.Iterator;
+import java.util.LinkedList;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.util.SafeRun;
import org.apache.pulsar.broker.systopic.NamespaceEventsSystemTopicFactory;
import org.apache.pulsar.broker.systopic.SystemTopicClient;
import org.apache.pulsar.broker.systopic.SystemTopicClient.Reader;
import org.apache.pulsar.broker.systopic.SystemTopicClient.Writer;
import org.apache.pulsar.broker.systopic.TransactionBufferSystemTopicClient;
import org.apache.pulsar.broker.transaction.buffer.matadata.TransactionBufferSnapshot;
import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.api.PulsarClientException.InvalidTopicNameException;
-import org.apache.pulsar.common.events.EventType;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.Backoff;
+import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.util.FutureUtil;
+@Slf4j
public class SystemTopicBaseTxnBufferSnapshotService implements TransactionBufferSnapshotService {
- private final Map<TopicName, SystemTopicClient<TransactionBufferSnapshot>> clients;
+ private final Map<NamespaceName, SystemTopicClient<TransactionBufferSnapshot>> clients;
private final NamespaceEventsSystemTopicFactory namespaceEventsSystemTopicFactory;
- public SystemTopicBaseTxnBufferSnapshotService(PulsarClient client) {
+ private final ScheduledExecutorService scheduledExecutorService;
+ private final ConcurrentHashMap<NamespaceName, ReferenceCountedWriter> writerFutureMap;
+ private final LinkedList<CompletableFuture<Writer<TransactionBufferSnapshot>>> pendingCloseWriterList;
+
+ // The class ReferenceCountedWriter will maintain the reference count,
+ // when the reference count decrement to 0, it will be removed from writerFutureMap, the writer will be closed.
+ public static class ReferenceCountedWriter extends AbstractReferenceCounted {
+
+ private final NamespaceName namespaceName;
+ private final SystemTopicBaseTxnBufferSnapshotService service;
+ private CompletableFuture<Writer<TransactionBufferSnapshot>> future;
+ private final Backoff backoff;
+
+ protected ReferenceCountedWriter(NamespaceName namespaceName,
+ SystemTopicBaseTxnBufferSnapshotService service) {
+ this.namespaceName = namespaceName;
+ this.service = service;
+ this.backoff = new Backoff(1, TimeUnit.SECONDS, 3, TimeUnit.SECONDS, 10, TimeUnit.SECONDS);
+ initWriterFuture();
+ }
+
+ private synchronized void initWriterFuture() {
+ this.future = service.getTransactionBufferSystemTopicClient(namespaceName).newWriterAsync();
+ this.future.thenRunAsync(this.backoff::reset).exceptionally(throwable -> {
Review Comment:
Why use ``thenRunAsync`` with `common-pool `here? I'm not sure we should use a specific executor?
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBaseTxnBufferSnapshotService.java:
##########
@@ -18,69 +18,176 @@
*/
package org.apache.pulsar.broker.service;
+import io.netty.util.AbstractReferenceCounted;
+import io.netty.util.ReferenceCounted;
+import java.util.Iterator;
+import java.util.LinkedList;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.util.SafeRun;
import org.apache.pulsar.broker.systopic.NamespaceEventsSystemTopicFactory;
import org.apache.pulsar.broker.systopic.SystemTopicClient;
import org.apache.pulsar.broker.systopic.SystemTopicClient.Reader;
import org.apache.pulsar.broker.systopic.SystemTopicClient.Writer;
import org.apache.pulsar.broker.systopic.TransactionBufferSystemTopicClient;
import org.apache.pulsar.broker.transaction.buffer.matadata.TransactionBufferSnapshot;
import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.api.PulsarClientException.InvalidTopicNameException;
-import org.apache.pulsar.common.events.EventType;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.Backoff;
+import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.util.FutureUtil;
+@Slf4j
public class SystemTopicBaseTxnBufferSnapshotService implements TransactionBufferSnapshotService {
- private final Map<TopicName, SystemTopicClient<TransactionBufferSnapshot>> clients;
+ private final Map<NamespaceName, SystemTopicClient<TransactionBufferSnapshot>> clients;
private final NamespaceEventsSystemTopicFactory namespaceEventsSystemTopicFactory;
- public SystemTopicBaseTxnBufferSnapshotService(PulsarClient client) {
+ private final ScheduledExecutorService scheduledExecutorService;
+ private final ConcurrentHashMap<NamespaceName, ReferenceCountedWriter> writerFutureMap;
+ private final LinkedList<CompletableFuture<Writer<TransactionBufferSnapshot>>> pendingCloseWriterList;
+
+ // The class ReferenceCountedWriter will maintain the reference count,
+ // when the reference count decrement to 0, it will be removed from writerFutureMap, the writer will be closed.
+ public static class ReferenceCountedWriter extends AbstractReferenceCounted {
+
+ private final NamespaceName namespaceName;
+ private final SystemTopicBaseTxnBufferSnapshotService service;
+ private CompletableFuture<Writer<TransactionBufferSnapshot>> future;
+ private final Backoff backoff;
+
+ protected ReferenceCountedWriter(NamespaceName namespaceName,
+ SystemTopicBaseTxnBufferSnapshotService service) {
+ this.namespaceName = namespaceName;
+ this.service = service;
+ this.backoff = new Backoff(1, TimeUnit.SECONDS, 3, TimeUnit.SECONDS, 10, TimeUnit.SECONDS);
+ initWriterFuture();
+ }
+
+ private synchronized void initWriterFuture() {
+ this.future = service.getTransactionBufferSystemTopicClient(namespaceName).newWriterAsync();
+ this.future.thenRunAsync(this.backoff::reset).exceptionally(throwable -> {
+ long delay = backoff.next();
+ log.error("[{}] Failed to new transaction buffer system topic writer,"
+ + "try to re-create the writer in {} ms.", delay, namespaceName, throwable);
+ service.scheduledExecutorService.schedule(
+ SafeRun.safeRun(this::initWriterFuture), delay, TimeUnit.MILLISECONDS);
+ return null;
+ });
+ }
+
+ public CompletableFuture<Writer<TransactionBufferSnapshot>> getFuture() {
+ if (future == null) {
+ // normally, this will not happen, not affect reference count, only avoid return a null object.
+ initWriterFuture();
+ }
+ return future;
+ }
+
+ @Override
+ protected void deallocate() {
+ service.writerFutureMap.compute(namespaceName, (k, v) -> {
+ if (v == this) {
+ // only remove it's self, avoid remove new add reference count object
+ service.writerFutureMap.remove(namespaceName);
+ service.pendingCloseWriterList.add(this.future);
+ service.closePendingCloseWriter();
+ return null;
+ }
+ return v;
+ });
+ }
+
+ @Override
+ public ReferenceCounted touch(Object o) {
+ return this;
+ }
+
+ }
+
+ public SystemTopicBaseTxnBufferSnapshotService(PulsarClient client,
+ ScheduledExecutorService scheduledExecutorService) {
this.namespaceEventsSystemTopicFactory = new NamespaceEventsSystemTopicFactory(client);
this.clients = new ConcurrentHashMap<>();
+ this.scheduledExecutorService = scheduledExecutorService;
+ this.writerFutureMap = new ConcurrentHashMap<>();
+ this.pendingCloseWriterList = new LinkedList<>();
}
@Override
public CompletableFuture<Writer<TransactionBufferSnapshot>> createWriter(TopicName topicName) {
- return getTransactionBufferSystemTopicClient(topicName).thenCompose(SystemTopicClient::newWriterAsync);
- }
-
- private CompletableFuture<SystemTopicClient<TransactionBufferSnapshot>> getTransactionBufferSystemTopicClient(
- TopicName topicName) {
- TopicName systemTopicName = NamespaceEventsSystemTopicFactory
- .getSystemTopicName(topicName.getNamespaceObject(), EventType.TRANSACTION_BUFFER_SNAPSHOT);
- if (systemTopicName == null) {
+ if (topicName == null) {
return FutureUtil.failedFuture(
- new InvalidTopicNameException("Can't create SystemTopicBaseTxnBufferSnapshotService, "
- + "because the topicName is null!"));
+ new PulsarClientException.InvalidTopicNameException(
+ "Can't create SystemTopicBaseTxnBufferSnapshotService, because the topicName is null!"));
}
- return CompletableFuture.completedFuture(clients.computeIfAbsent(systemTopicName,
+ return getTransactionBufferSystemTopicClient(topicName.getNamespaceObject()).newWriterAsync();
+ }
+
+ @Override
+ public ReferenceCountedWriter createReferenceWriter(NamespaceName namespaceName) {
+ return writerFutureMap.compute(namespaceName, (ns, writerFuture) -> {
+ if (writerFuture == null) {
+ return new ReferenceCountedWriter(namespaceName, this);
+ }
+ try {
+ writerFuture.retain();
+ } catch (Exception e) {
+ // Resolve potential race condition problem, if retain method encounter reference count exception
+ // or other exceptions, create a new `ReferenceCountedWriter`, when the `ReferenceCountedWriter` release
+ // but didn't remove from `writerFutureMap`.
+ return new ReferenceCountedWriter(namespaceName, this);
Review Comment:
Question: Should we close the original writer?
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBaseTxnBufferSnapshotService.java:
##########
@@ -18,69 +18,176 @@
*/
package org.apache.pulsar.broker.service;
+import io.netty.util.AbstractReferenceCounted;
+import io.netty.util.ReferenceCounted;
+import java.util.Iterator;
+import java.util.LinkedList;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.util.SafeRun;
import org.apache.pulsar.broker.systopic.NamespaceEventsSystemTopicFactory;
import org.apache.pulsar.broker.systopic.SystemTopicClient;
import org.apache.pulsar.broker.systopic.SystemTopicClient.Reader;
import org.apache.pulsar.broker.systopic.SystemTopicClient.Writer;
import org.apache.pulsar.broker.systopic.TransactionBufferSystemTopicClient;
import org.apache.pulsar.broker.transaction.buffer.matadata.TransactionBufferSnapshot;
import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.api.PulsarClientException.InvalidTopicNameException;
-import org.apache.pulsar.common.events.EventType;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.Backoff;
+import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.util.FutureUtil;
+@Slf4j
public class SystemTopicBaseTxnBufferSnapshotService implements TransactionBufferSnapshotService {
- private final Map<TopicName, SystemTopicClient<TransactionBufferSnapshot>> clients;
+ private final Map<NamespaceName, SystemTopicClient<TransactionBufferSnapshot>> clients;
private final NamespaceEventsSystemTopicFactory namespaceEventsSystemTopicFactory;
- public SystemTopicBaseTxnBufferSnapshotService(PulsarClient client) {
+ private final ScheduledExecutorService scheduledExecutorService;
+ private final ConcurrentHashMap<NamespaceName, ReferenceCountedWriter> writerFutureMap;
+ private final LinkedList<CompletableFuture<Writer<TransactionBufferSnapshot>>> pendingCloseWriterList;
+
+ // The class ReferenceCountedWriter will maintain the reference count,
+ // when the reference count decrement to 0, it will be removed from writerFutureMap, the writer will be closed.
+ public static class ReferenceCountedWriter extends AbstractReferenceCounted {
+
+ private final NamespaceName namespaceName;
+ private final SystemTopicBaseTxnBufferSnapshotService service;
+ private CompletableFuture<Writer<TransactionBufferSnapshot>> future;
+ private final Backoff backoff;
+
+ protected ReferenceCountedWriter(NamespaceName namespaceName,
+ SystemTopicBaseTxnBufferSnapshotService service) {
+ this.namespaceName = namespaceName;
+ this.service = service;
+ this.backoff = new Backoff(1, TimeUnit.SECONDS, 3, TimeUnit.SECONDS, 10, TimeUnit.SECONDS);
+ initWriterFuture();
+ }
+
+ private synchronized void initWriterFuture() {
+ this.future = service.getTransactionBufferSystemTopicClient(namespaceName).newWriterAsync();
+ this.future.thenRunAsync(this.backoff::reset).exceptionally(throwable -> {
+ long delay = backoff.next();
+ log.error("[{}] Failed to new transaction buffer system topic writer,"
+ + "try to re-create the writer in {} ms.", delay, namespaceName, throwable);
+ service.scheduledExecutorService.schedule(
+ SafeRun.safeRun(this::initWriterFuture), delay, TimeUnit.MILLISECONDS);
+ return null;
+ });
+ }
+
+ public CompletableFuture<Writer<TransactionBufferSnapshot>> getFuture() {
+ if (future == null) {
+ // normally, this will not happen, not affect reference count, only avoid return a null object.
+ initWriterFuture();
+ }
+ return future;
+ }
+
+ @Override
+ protected void deallocate() {
+ service.writerFutureMap.compute(namespaceName, (k, v) -> {
+ if (v == this) {
+ // only remove it's self, avoid remove new add reference count object
+ service.writerFutureMap.remove(namespaceName);
+ service.pendingCloseWriterList.add(this.future);
+ service.closePendingCloseWriter();
+ return null;
+ }
+ return v;
+ });
+ }
+
+ @Override
+ public ReferenceCounted touch(Object o) {
+ return this;
+ }
+
+ }
+
+ public SystemTopicBaseTxnBufferSnapshotService(PulsarClient client,
+ ScheduledExecutorService scheduledExecutorService) {
this.namespaceEventsSystemTopicFactory = new NamespaceEventsSystemTopicFactory(client);
this.clients = new ConcurrentHashMap<>();
+ this.scheduledExecutorService = scheduledExecutorService;
+ this.writerFutureMap = new ConcurrentHashMap<>();
+ this.pendingCloseWriterList = new LinkedList<>();
}
@Override
public CompletableFuture<Writer<TransactionBufferSnapshot>> createWriter(TopicName topicName) {
- return getTransactionBufferSystemTopicClient(topicName).thenCompose(SystemTopicClient::newWriterAsync);
- }
-
- private CompletableFuture<SystemTopicClient<TransactionBufferSnapshot>> getTransactionBufferSystemTopicClient(
- TopicName topicName) {
- TopicName systemTopicName = NamespaceEventsSystemTopicFactory
- .getSystemTopicName(topicName.getNamespaceObject(), EventType.TRANSACTION_BUFFER_SNAPSHOT);
- if (systemTopicName == null) {
+ if (topicName == null) {
return FutureUtil.failedFuture(
- new InvalidTopicNameException("Can't create SystemTopicBaseTxnBufferSnapshotService, "
- + "because the topicName is null!"));
+ new PulsarClientException.InvalidTopicNameException(
+ "Can't create SystemTopicBaseTxnBufferSnapshotService, because the topicName is null!"));
}
- return CompletableFuture.completedFuture(clients.computeIfAbsent(systemTopicName,
+ return getTransactionBufferSystemTopicClient(topicName.getNamespaceObject()).newWriterAsync();
+ }
+
+ @Override
+ public ReferenceCountedWriter createReferenceWriter(NamespaceName namespaceName) {
+ return writerFutureMap.compute(namespaceName, (ns, writerFuture) -> {
+ if (writerFuture == null) {
+ return new ReferenceCountedWriter(namespaceName, this);
+ }
+ try {
+ writerFuture.retain();
+ } catch (Exception e) {
+ // Resolve potential race condition problem, if retain method encounter reference count exception
+ // or other exceptions, create a new `ReferenceCountedWriter`, when the `ReferenceCountedWriter` release
+ // but didn't remove from `writerFutureMap`.
+ return new ReferenceCountedWriter(namespaceName, this);
+ }
+ return writerFuture;
+ });
+ }
+
+ private SystemTopicClient<TransactionBufferSnapshot> getTransactionBufferSystemTopicClient(
+ NamespaceName namespaceName) {
+ return clients.computeIfAbsent(namespaceName,
(v) -> namespaceEventsSystemTopicFactory
- .createTransactionBufferSystemTopicClient(topicName.getNamespaceObject(), this)));
+ .createTransactionBufferSystemTopicClient(namespaceName, this));
}
@Override
public CompletableFuture<Reader<TransactionBufferSnapshot>> createReader(TopicName topicName) {
- return getTransactionBufferSystemTopicClient(topicName).thenCompose(SystemTopicClient::newReaderAsync);
+ return getTransactionBufferSystemTopicClient(topicName.getNamespaceObject()).newReaderAsync();
}
@Override
public void removeClient(TopicName topicName,
TransactionBufferSystemTopicClient transactionBufferSystemTopicClient) {
if (transactionBufferSystemTopicClient.getReaders().size() == 0
&& transactionBufferSystemTopicClient.getWriters().size() == 0) {
- clients.remove(topicName);
+ clients.remove(topicName.getNamespaceObject());
}
}
@Override
public void close() throws Exception {
- for (Map.Entry<TopicName, SystemTopicClient<TransactionBufferSnapshot>> entry : clients.entrySet()) {
+ for (Map.Entry<NamespaceName, SystemTopicClient<TransactionBufferSnapshot>> entry : clients.entrySet()) {
entry.getValue().close();
}
}
+
+ private void closePendingCloseWriter() {
+ Iterator<CompletableFuture<Writer<TransactionBufferSnapshot>>> iterator =
+ pendingCloseWriterList.stream().iterator();
+ while (iterator.hasNext()) {
+ CompletableFuture<Writer<TransactionBufferSnapshot>> future = iterator.next();
+ if (future == null) {
+ continue;
+ }
+ future.thenAccept(writer ->
+ writer.closeAsync().thenAccept(ignore ->
Review Comment:
When the close fails, ``thenAccept`` will not be called, and then the future will not be deleted.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBaseTxnBufferSnapshotService.java:
##########
@@ -18,69 +18,176 @@
*/
package org.apache.pulsar.broker.service;
+import io.netty.util.AbstractReferenceCounted;
+import io.netty.util.ReferenceCounted;
+import java.util.Iterator;
+import java.util.LinkedList;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.mledger.util.SafeRun;
import org.apache.pulsar.broker.systopic.NamespaceEventsSystemTopicFactory;
import org.apache.pulsar.broker.systopic.SystemTopicClient;
import org.apache.pulsar.broker.systopic.SystemTopicClient.Reader;
import org.apache.pulsar.broker.systopic.SystemTopicClient.Writer;
import org.apache.pulsar.broker.systopic.TransactionBufferSystemTopicClient;
import org.apache.pulsar.broker.transaction.buffer.matadata.TransactionBufferSnapshot;
import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.api.PulsarClientException.InvalidTopicNameException;
-import org.apache.pulsar.common.events.EventType;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.Backoff;
+import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.util.FutureUtil;
+@Slf4j
public class SystemTopicBaseTxnBufferSnapshotService implements TransactionBufferSnapshotService {
- private final Map<TopicName, SystemTopicClient<TransactionBufferSnapshot>> clients;
+ private final Map<NamespaceName, SystemTopicClient<TransactionBufferSnapshot>> clients;
private final NamespaceEventsSystemTopicFactory namespaceEventsSystemTopicFactory;
- public SystemTopicBaseTxnBufferSnapshotService(PulsarClient client) {
+ private final ScheduledExecutorService scheduledExecutorService;
+ private final ConcurrentHashMap<NamespaceName, ReferenceCountedWriter> writerFutureMap;
+ private final LinkedList<CompletableFuture<Writer<TransactionBufferSnapshot>>> pendingCloseWriterList;
+
+ // The class ReferenceCountedWriter will maintain the reference count,
+ // when the reference count decrement to 0, it will be removed from writerFutureMap, the writer will be closed.
+ public static class ReferenceCountedWriter extends AbstractReferenceCounted {
+
+ private final NamespaceName namespaceName;
+ private final SystemTopicBaseTxnBufferSnapshotService service;
+ private CompletableFuture<Writer<TransactionBufferSnapshot>> future;
+ private final Backoff backoff;
+
+ protected ReferenceCountedWriter(NamespaceName namespaceName,
+ SystemTopicBaseTxnBufferSnapshotService service) {
+ this.namespaceName = namespaceName;
+ this.service = service;
+ this.backoff = new Backoff(1, TimeUnit.SECONDS, 3, TimeUnit.SECONDS, 10, TimeUnit.SECONDS);
+ initWriterFuture();
+ }
+
+ private synchronized void initWriterFuture() {
+ this.future = service.getTransactionBufferSystemTopicClient(namespaceName).newWriterAsync();
+ this.future.thenRunAsync(this.backoff::reset).exceptionally(throwable -> {
+ long delay = backoff.next();
+ log.error("[{}] Failed to new transaction buffer system topic writer,"
+ + "try to re-create the writer in {} ms.", delay, namespaceName, throwable);
+ service.scheduledExecutorService.schedule(
+ SafeRun.safeRun(this::initWriterFuture), delay, TimeUnit.MILLISECONDS);
+ return null;
+ });
+ }
+
+ public CompletableFuture<Writer<TransactionBufferSnapshot>> getFuture() {
+ if (future == null) {
+ // normally, this will not happen, not affect reference count, only avoid return a null object.
+ initWriterFuture();
+ }
+ return future;
+ }
+
+ @Override
+ protected void deallocate() {
+ service.writerFutureMap.compute(namespaceName, (k, v) -> {
+ if (v == this) {
+ // only remove it's self, avoid remove new add reference count object
+ service.writerFutureMap.remove(namespaceName);
+ service.pendingCloseWriterList.add(this.future);
+ service.closePendingCloseWriter();
+ return null;
+ }
+ return v;
+ });
+ }
+
+ @Override
+ public ReferenceCounted touch(Object o) {
+ return this;
+ }
+
+ }
+
+ public SystemTopicBaseTxnBufferSnapshotService(PulsarClient client,
+ ScheduledExecutorService scheduledExecutorService) {
this.namespaceEventsSystemTopicFactory = new NamespaceEventsSystemTopicFactory(client);
this.clients = new ConcurrentHashMap<>();
+ this.scheduledExecutorService = scheduledExecutorService;
+ this.writerFutureMap = new ConcurrentHashMap<>();
+ this.pendingCloseWriterList = new LinkedList<>();
}
@Override
public CompletableFuture<Writer<TransactionBufferSnapshot>> createWriter(TopicName topicName) {
- return getTransactionBufferSystemTopicClient(topicName).thenCompose(SystemTopicClient::newWriterAsync);
- }
-
- private CompletableFuture<SystemTopicClient<TransactionBufferSnapshot>> getTransactionBufferSystemTopicClient(
- TopicName topicName) {
- TopicName systemTopicName = NamespaceEventsSystemTopicFactory
- .getSystemTopicName(topicName.getNamespaceObject(), EventType.TRANSACTION_BUFFER_SNAPSHOT);
- if (systemTopicName == null) {
+ if (topicName == null) {
return FutureUtil.failedFuture(
- new InvalidTopicNameException("Can't create SystemTopicBaseTxnBufferSnapshotService, "
- + "because the topicName is null!"));
+ new PulsarClientException.InvalidTopicNameException(
+ "Can't create SystemTopicBaseTxnBufferSnapshotService, because the topicName is null!"));
}
- return CompletableFuture.completedFuture(clients.computeIfAbsent(systemTopicName,
+ return getTransactionBufferSystemTopicClient(topicName.getNamespaceObject()).newWriterAsync();
+ }
+
+ @Override
+ public ReferenceCountedWriter createReferenceWriter(NamespaceName namespaceName) {
+ return writerFutureMap.compute(namespaceName, (ns, writerFuture) -> {
+ if (writerFuture == null) {
+ return new ReferenceCountedWriter(namespaceName, this);
+ }
+ try {
+ writerFuture.retain();
+ } catch (Exception e) {
+ // Resolve potential race condition problem, if retain method encounter reference count exception
+ // or other exceptions, create a new `ReferenceCountedWriter`, when the `ReferenceCountedWriter` release
+ // but didn't remove from `writerFutureMap`.
+ return new ReferenceCountedWriter(namespaceName, this);
+ }
+ return writerFuture;
+ });
+ }
+
+ private SystemTopicClient<TransactionBufferSnapshot> getTransactionBufferSystemTopicClient(
+ NamespaceName namespaceName) {
+ return clients.computeIfAbsent(namespaceName,
(v) -> namespaceEventsSystemTopicFactory
- .createTransactionBufferSystemTopicClient(topicName.getNamespaceObject(), this)));
+ .createTransactionBufferSystemTopicClient(namespaceName, this));
}
@Override
public CompletableFuture<Reader<TransactionBufferSnapshot>> createReader(TopicName topicName) {
- return getTransactionBufferSystemTopicClient(topicName).thenCompose(SystemTopicClient::newReaderAsync);
+ return getTransactionBufferSystemTopicClient(topicName.getNamespaceObject()).newReaderAsync();
}
@Override
public void removeClient(TopicName topicName,
TransactionBufferSystemTopicClient transactionBufferSystemTopicClient) {
if (transactionBufferSystemTopicClient.getReaders().size() == 0
&& transactionBufferSystemTopicClient.getWriters().size() == 0) {
- clients.remove(topicName);
+ clients.remove(topicName.getNamespaceObject());
}
}
@Override
public void close() throws Exception {
- for (Map.Entry<TopicName, SystemTopicClient<TransactionBufferSnapshot>> entry : clients.entrySet()) {
+ for (Map.Entry<NamespaceName, SystemTopicClient<TransactionBufferSnapshot>> entry : clients.entrySet()) {
entry.getValue().close();
}
}
+
+ private void closePendingCloseWriter() {
+ Iterator<CompletableFuture<Writer<TransactionBufferSnapshot>>> iterator =
+ pendingCloseWriterList.stream().iterator();
+ while (iterator.hasNext()) {
+ CompletableFuture<Writer<TransactionBufferSnapshot>> future = iterator.next();
+ if (future == null) {
Review Comment:
Question: Why is iterator `hasNext` still have to judge whether the future exists?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org