From 8b7195170f106c623a050065cfa6f17140a27a0e Mon Sep 17 00:00:00 2001 From: Sijie Guo Date: Mon, 20 Jan 2020 14:26:10 -0800 Subject: [PATCH 1/3] [Functions] The argument and description for dead letter topic is wrong *Motivation* #5400 introduces `customRuntimeOptions` in function details. But the description was wrong. The mistake was probably introduced by bad merges. *Modification* Fix the argument and description for `deadletterTopic` and `customRuntimeOptions`. *Tests* Add a unit test to ensure the parameters are set correctly. --- .../java/org/apache/pulsar/admin/cli/CmdFunctionsTest.java | 4 ++++ .../main/java/org/apache/pulsar/admin/cli/CmdFunctions.java | 4 ++-- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/CmdFunctionsTest.java b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/CmdFunctionsTest.java index d9bc24888936c..04d833b3ffc74 100644 --- a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/CmdFunctionsTest.java +++ b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/CmdFunctionsTest.java @@ -185,6 +185,8 @@ public void testCreateFunction() throws Exception { "--tenant", "sample", "--namespace", "ns1", "--className", DummyFunction.class.getName(), + "--dead-letter-topic", "test-dead-letter-topic", + "--custom-runtime-options", "custom-runtime-options" }); CreateFunction creater = cmd.getCreater(); @@ -192,6 +194,8 @@ public void testCreateFunction() throws Exception { assertEquals(inputTopicName, creater.getInputs()); assertEquals(outputTopicName, creater.getOutput()); assertEquals(new Boolean(false), creater.getAutoAck()); + assertEquals("test-dead-letter-topic", creater.getDeadLetterTopic()); + assertEquals("custom-runtime-options", creater.getCustomRuntimeOptions()); verify(functions, times(1)).createFunction(any(FunctionConfig.class), anyString()); diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java index d305301fd6f4d..2787bc322c456 100644 --- a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java +++ b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java @@ -299,9 +299,9 @@ abstract class FunctionDetailsCommand extends BaseCommand { protected Long timeoutMs; @Parameter(names = "--max-message-retries", description = "How many times should we try to process a message before giving up") protected Integer maxMessageRetries; - @Parameter(names = "--dead-letter-topic", description = "The topic where messages that are not processed successfully are sent to") - protected String customRuntimeOptions; @Parameter(names = "--custom-runtime-options", description = "A string that encodes options to customize the runtime, see docs for configured runtime for details") + protected String customRuntimeOptions; + @Parameter(names = "--dead-letter-topic", description = "The topic where messages that are not processed successfully are sent to") protected String deadLetterTopic; protected FunctionConfig functionConfig; protected String userCodeFile; From 0e479065e87e5130f4beca535af155c8b7c885d2 Mon Sep 17 00:00:00 2001 From: Sijie Guo Date: Mon, 20 Jan 2020 17:52:51 -0800 Subject: [PATCH 2/3] [Websocket] Websocket doesn't set the correct cluster data *Motivation* A regression was introduced in #5486. If websocket service as running as part of pulsar standalone, the cluster data is set with null service urls. This causes service url is not set correctly in the pulsar client and an illegal argument exception ("Param serviceUrl must not be blank.") will be thrown. *Modifications* 1. Pass `null` when constructing the websocket service. So the local cluster data can be refreshed when creating pulsar client. 2. Set the cluster data after both broker service and web service started and ports are allocated. *Test* Verified locally and need to figure out how to automate this in integration tests --- .../main/java/org/apache/pulsar/broker/PulsarService.java | 8 +++++--- .../org/apache/pulsar/websocket/WebSocketService.java | 2 ++ 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java index fd169fe1d1d59..65651d605af4d 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java @@ -427,9 +427,7 @@ public Boolean get() { if (config.isWebSocketServiceEnabled()) { // Use local broker address to avoid different IP address when using a VIP for service discovery - this.webSocketService = new WebSocketService( - new ClusterData(webServiceAddress, webServiceAddressTls, brokerServiceUrl, brokerServiceUrlTls), - config); + this.webSocketService = new WebSocketService(null, config); this.webSocketService.start(); final WebSocketServlet producerWebSocketServlet = new WebSocketProducerServlet(webSocketService); @@ -466,6 +464,10 @@ public Boolean get() { this.brokerServiceUrl = brokerUrl(config); this.brokerServiceUrlTls = brokerUrlTls(config); + ClusterData clusterData = + new ClusterData(webServiceAddress, webServiceAddressTls, brokerServiceUrl, brokerServiceUrlTls); + this.webSocketService.setLocalCluster(clusterData); + // needs load management service this.startNamespaceService(); diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java index 79cd4366e9fdf..062af7efb82ed 100644 --- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java +++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/WebSocketService.java @@ -30,6 +30,7 @@ import javax.servlet.ServletException; import javax.websocket.DeploymentException; +import lombok.Setter; import org.apache.bookkeeper.common.util.OrderedScheduler; import org.apache.pulsar.broker.PulsarServerException; import org.apache.pulsar.broker.ServiceConfiguration; @@ -77,6 +78,7 @@ public class WebSocketService implements Closeable { private ServiceConfiguration config; private ConfigurationCacheService configurationCacheService; + @Setter private ClusterData localCluster; private final ConcurrentOpenHashMap> topicProducerMap; private final ConcurrentOpenHashMap> topicConsumerMap; From adc9ccb4fba30e38212e21f34952613c0ca3fea6 Mon Sep 17 00:00:00 2001 From: Sijie Guo Date: Wed, 22 Jan 2020 01:28:21 -0800 Subject: [PATCH 3/3] Check if webSocketService is set --- .../main/java/org/apache/pulsar/broker/PulsarService.java | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java index 65651d605af4d..ff7ba47728a9c 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java @@ -464,9 +464,11 @@ public Boolean get() { this.brokerServiceUrl = brokerUrl(config); this.brokerServiceUrlTls = brokerUrlTls(config); - ClusterData clusterData = - new ClusterData(webServiceAddress, webServiceAddressTls, brokerServiceUrl, brokerServiceUrlTls); - this.webSocketService.setLocalCluster(clusterData); + if (null != this.webSocketService) { + ClusterData clusterData = + new ClusterData(webServiceAddress, webServiceAddressTls, brokerServiceUrl, brokerServiceUrlTls); + this.webSocketService.setLocalCluster(clusterData); + } // needs load management service this.startNamespaceService();