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 2021/08/04 03:05:45 UTC
[GitHub] [pulsar] zymap commented on a change in pull request #11543: Donot create transaction components for function work topic
zymap commented on a change in pull request #11543:
URL: https://github.com/apache/pulsar/pull/11543#discussion_r682248311
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
##########
@@ -149,13 +153,30 @@ public PersistentSubscription(PersistentTopic topic, String subscriptionName, Ma
this.expiryMonitor = new PersistentMessageExpiryMonitor(topicName, subscriptionName, cursor, this);
this.setReplicated(replicated);
if (topic.getBrokerService().getPulsar().getConfig().isTransactionCoordinatorEnabled()
- && !checkTopicIsEventsNames(TopicName.get(topicName))) {
+ && !checkTopicIsEventsNames(TopicName.get(topicName))
+ && !checkTopicIsFunctionWorkerService(topic)) {
this.pendingAckHandle = new PendingAckHandleImpl(this);
} else {
this.pendingAckHandle = new PendingAckHandleDisabled();
}
IS_FENCED_UPDATER.set(this, FALSE);
}
+ private boolean checkTopicIsFunctionWorkerService(PersistentTopic topic){
+ String fnWorkerConfigFile =
+ Paths.get("").toAbsolutePath().normalize().toString() + "/conf/functions_worker.yml";
+ WorkerConfig workerConfig = null;
+ try {
+ workerConfig = WorkerConfig.load(fnWorkerConfigFile);
+ } catch (IOException e) {
+ e.printStackTrace();
Review comment:
please handle the exception not print it
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
##########
@@ -149,13 +153,30 @@ public PersistentSubscription(PersistentTopic topic, String subscriptionName, Ma
this.expiryMonitor = new PersistentMessageExpiryMonitor(topicName, subscriptionName, cursor, this);
this.setReplicated(replicated);
if (topic.getBrokerService().getPulsar().getConfig().isTransactionCoordinatorEnabled()
- && !checkTopicIsEventsNames(TopicName.get(topicName))) {
+ && !checkTopicIsEventsNames(TopicName.get(topicName))
+ && !checkTopicIsFunctionWorkerService(topic)) {
this.pendingAckHandle = new PendingAckHandleImpl(this);
} else {
this.pendingAckHandle = new PendingAckHandleDisabled();
}
IS_FENCED_UPDATER.set(this, FALSE);
}
+ private boolean checkTopicIsFunctionWorkerService(PersistentTopic topic){
+ String fnWorkerConfigFile =
+ Paths.get("").toAbsolutePath().normalize().toString() + "/conf/functions_worker.yml";
+ WorkerConfig workerConfig = null;
+ try {
+ workerConfig = WorkerConfig.load(fnWorkerConfigFile);
+ } catch (IOException e) {
+ e.printStackTrace();
+ }
+ Assert.isNonEmpty(workerConfig);
+ Assert.isNonEmpty(topic);
Review comment:
Don't we need to handle the returned value?
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
##########
@@ -149,13 +153,30 @@ public PersistentSubscription(PersistentTopic topic, String subscriptionName, Ma
this.expiryMonitor = new PersistentMessageExpiryMonitor(topicName, subscriptionName, cursor, this);
this.setReplicated(replicated);
if (topic.getBrokerService().getPulsar().getConfig().isTransactionCoordinatorEnabled()
- && !checkTopicIsEventsNames(TopicName.get(topicName))) {
+ && !checkTopicIsEventsNames(TopicName.get(topicName))
+ && !checkTopicIsFunctionWorkerService(topic)) {
this.pendingAckHandle = new PendingAckHandleImpl(this);
} else {
this.pendingAckHandle = new PendingAckHandleDisabled();
}
IS_FENCED_UPDATER.set(this, FALSE);
}
+ private boolean checkTopicIsFunctionWorkerService(PersistentTopic topic){
+ String fnWorkerConfigFile =
Review comment:
The function work config file may be in a different place, user can define it with environment variable
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
##########
@@ -300,6 +305,22 @@ public PersistentTopic(String topic, ManagedLedger ledger, BrokerService brokerS
}
transactionBuffer.syncMaxReadPositionForNormalPublish((PositionImpl) ledger.getLastConfirmedEntry());
}
+ private boolean checkTopicIsFunctionWorkerService(PersistentTopic topic){
Review comment:
We'd better find a common place to avoid the method duplicate
--
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