sijie commented on a change in pull request #5680: [transaction-coordinator]
Implementation of transaction coordinator client.
URL: https://github.com/apache/pulsar/pull/5680#discussion_r348704078
##########
File path:
pulsar-broker/src/main/java/org/apache/pulsar/broker/TransactionMetadataStoreService.java
##########
@@ -50,29 +55,80 @@ public
TransactionMetadataStoreService(TransactionMetadataStoreProvider transact
this.transactionMetadataStoreProvider =
transactionMetadataStoreProvider;
}
+ public void start() {
+
pulsarService.getNamespaceService().addNamespaceBundleOwnershipListener(new
NamespaceBundleOwnershipListener() {
+ @Override
+ public void onLoad(NamespaceBundle bundle) {
+
pulsarService.getNamespaceService().getOwnedTopicListForNamespaceBundle(bundle)
+ .whenComplete((topics, ex) -> {
+ if (ex == null) {
+ for (String topic : topics) {
+ TopicName name = TopicName.get(topic);
+ if
(TopicName.TRANSACTION_COORDINATOR_ASSIGN.getLocalName()
+
.equals(TopicName.get(name.getPartitionedTopicName()).getLocalName())
+ && name.isPartitioned()) {
+
addTransactionMetadataStore(TransactionCoordinatorID.get(name.getPartitionIndex()));
+ }
+ }
+ } else {
+ LOG.error("Get owned topic list error when trigger
bundle {} onload.", bundle, ex);
+ }
+ });
+ }
+ @Override
+ public void unLoad(NamespaceBundle bundle) {
+
pulsarService.getNamespaceService().getOwnedTopicListForNamespaceBundle(bundle)
+ .whenComplete((topics, ex) -> {
+ if (ex == null) {
+ for (String topic : topics) {
+ TopicName name = TopicName.get(topic);
+ if
(TopicName.TRANSACTION_COORDINATOR_ASSIGN.getLocalName()
+
.equals(TopicName.get(name.getPartitionedTopicName()).getLocalName())
+ && name.isPartitioned()) {
+
removeTransactionMetadataStore(TransactionCoordinatorID.get(name.getPartitionIndex()));
+ }
+ }
+ } else {
+ LOG.error("Get owned topic list error when trigger
bundle {} unLoad.", bundle, ex);
Review comment:
```suggestion
LOG.error("Failed to get owned topic list error
when triggering un-loading bundle {}.", bundle, ex);
```
----------------------------------------------------------------
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]
With regards,
Apache Git Services
