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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
