poorbarcode commented on code in PR #16931:
URL: https://github.com/apache/pulsar/pull/16931#discussion_r999245551
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicTxnBufferSnapshotService.java:
##########
@@ -23,64 +23,63 @@
import java.util.concurrent.ConcurrentHashMap;
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.broker.systopic.SystemTopicClientBase;
import org.apache.pulsar.client.api.PulsarClient;
-import
org.apache.pulsar.client.api.PulsarClientException.InvalidTopicNameException;
+import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.common.events.EventType;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.util.FutureUtil;
-public class SystemTopicBaseTxnBufferSnapshotService implements
TransactionBufferSnapshotService {
+public class SystemTopicTxnBufferSnapshotService<T> {
- private final Map<TopicName, SystemTopicClient<TransactionBufferSnapshot>>
clients;
+ protected final Map<TopicName, SystemTopicClient<T>> clients;
+ protected final NamespaceEventsSystemTopicFactory
namespaceEventsSystemTopicFactory;
- private final NamespaceEventsSystemTopicFactory
namespaceEventsSystemTopicFactory;
+ protected final Class<T> schemaType;
+ protected final EventType systemTopicType;
- public SystemTopicBaseTxnBufferSnapshotService(PulsarClient client) {
+ public SystemTopicTxnBufferSnapshotService(PulsarClient client, EventType
systemTopicType,
+ Class<T> schemaType) {
this.namespaceEventsSystemTopicFactory = new
NamespaceEventsSystemTopicFactory(client);
+ this.systemTopicType = systemTopicType;
+ this.schemaType = schemaType;
this.clients = new ConcurrentHashMap<>();
}
- @Override
- public CompletableFuture<Writer<TransactionBufferSnapshot>>
createWriter(TopicName topicName) {
+ public CompletableFuture<SystemTopicClient.Writer<T>>
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) {
- return FutureUtil.failedFuture(
- new InvalidTopicNameException("Can't create
SystemTopicBaseTxnBufferSnapshotService, "
- + "because the topicName is null!"));
- }
- return
CompletableFuture.completedFuture(clients.computeIfAbsent(systemTopicName,
- (v) -> namespaceEventsSystemTopicFactory
-
.createTransactionBufferSystemTopicClient(topicName.getNamespaceObject(),
this)));
- }
-
- @Override
- public CompletableFuture<Reader<TransactionBufferSnapshot>>
createReader(TopicName topicName) {
+ public CompletableFuture<SystemTopicClient.Reader<T>>
createReader(TopicName topicName) {
return
getTransactionBufferSystemTopicClient(topicName).thenCompose(SystemTopicClient::newReaderAsync);
}
- @Override
- public void removeClient(TopicName topicName,
- TransactionBufferSystemTopicClient
transactionBufferSystemTopicClient) {
+ public void removeClient(TopicName topicName, SystemTopicClientBase<T>
transactionBufferSystemTopicClient) {
if (transactionBufferSystemTopicClient.getReaders().size() == 0
&& transactionBufferSystemTopicClient.getWriters().size() ==
0) {
clients.remove(topicName);
Review Comment:
This class is very similar to the factory class used to register the
`SystemTopicClient` for system topics. There are many System topics under each
namespace. I think these resources should be released( `gc(SystemTopicClient)`
) on the namespace deleted event. Now such an implementation will have a race
condition:
- close the last `writer`
- registry a new `writer`
| close the last `writer` | registry a new `writer` |
|---|---|
| check `transactionBufferSystemTopicClient.getWriters().size() == 0` :
`true` | |
| | registry new one |
| remove(topic) | |
@liangyepianzhou and I have discussed and would like to consult your opinion
@codelipenghui
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
