This message was deleted.

Hello everyone. I am trying to confirm if Effectively Once is working on pulsar function. I had a simple pulsar function which like shown below, read from a single topic 'x' and output to 'y' topic

@Override

public String process(byte[] bytes, Context context) {

return "output-: " + new String(bytes);

}

Then I had another 2 separated programs • producer program: which produces message to 'x' topic • consumer program: which consumes message from 'y' topic I am using FunctionLocalRunner(2.8.0).

FunctionConfig functionConfig = new FunctionConfig();

functionConfig.setName("testing");

functionConfig.setInputs(Collections.singletonList("x"));

functionConfig.setClassName(TestEffectivelyOnce.class.getName());

functionConfig.setRuntime(FunctionConfig.Runtime.JAVA);

functionConfig.setProcessingGuarantees(FunctionConfig.ProcessingGuarantees.EFFECTIVELY_ONCE);

functionConfig.setOutput("y");

LocalRunner localRunner = LocalRunner.builder()

.functionConfig(functionConfig)

.brokerServiceUrl("pulsar://localhost:6650")

.build();

localRunner.start(true);

While producer-program is producing the messages, I restart the FunctionLocalRunner. Result: judging from the log from consumer-program, there are duplicated messages. • duplicated message:

msg-1224

msg-1225

msg-1226

msg-1227

....

Did I do something wrong or pulsar functions doesn't support this?

This message was deleted.

Hi guys, i was trying to deploy a python class that inherits from pulsar Function under a pulsar cluster env. However, the python class was recognized as a simple python function and got this error. Any thoughts?

This message was deleted.

This message was deleted.

This message was deleted.

Hello Team. We're facing an issue with ZK outages. The current ZK cluster we got has 5 nodes, so as per the 2N+1 rule, it should be able to withstand an outage of 2 nodes. I'm trying to introduce outages while the usual Pulsar operations (using OpenMessaging for producing and consuming from 10 topics(each 10 partitions) with a produce rate of 1000 msgs/s) being carried out. Two scenarios were tested: 1. Outages happening before the traffic is started: The broker and bookies detect the missing ZK nodes but the cluster reaches stability after a few minutes. However I can see logs on the broker about not able to create topic reader on the _change_events_ topic:

2. Outages happening during ongoing traffic. The pulsar client gets disconnected, keeps on trying to connect and keeps on failing. All traffic gets frozen and as per the broker and bookie logs, cluster stability is not regained and the below logs are seen endlessly on the broker:

This message was deleted.

This message was deleted.

This message was deleted.

This message was deleted.

This message was deleted.

This message was deleted.

This message was deleted.

This message was deleted.

This message was deleted.

This message was deleted.

Hi, i had pulsar install on AKS cluster long back by using following command.

helm install pulsar apache/pulsar --version 2.7.2 --set initialize=true --set tls.enabled=true -n pulsar

I'm trying to use same process again to install pulsar in AKS cluster, but getting Errors

kubectl get pods -n pulsar
NAME                                     READY   STATUS             RESTARTS   AGE
pulsar-bookie-0                          1/1     Running            0          4d21h
pulsar-bookie-1                          1/1     Running            0          4d21h
pulsar-bookie-2                          1/1     Running            0          4d21h
pulsar-bookie-3                          1/1     Running            0          4d21h
pulsar-bookie-init-nn4zl                 0/1     Completed          0          4d21h
pulsar-broker-0                          0/1     CrashLoopBackOff   1221       4d9h
pulsar-broker-1                          0/1     CrashLoopBackOff   1158       4d9h
pulsar-broker-2                          0/1     CrashLoopBackOff   1201       4d9h
pulsar-grafana-99b4976f7-25jss           1/1     Running            0          4d21h
pulsar-prometheus-5f5fb9978b-l7rw2       1/1     Running            0          4d21h
pulsar-proxy-0                           1/1     Running            0          4d21h
pulsar-proxy-1                           1/1     Running            0          4d21h
pulsar-proxy-2                           1/1     Running            0          4d21h
pulsar-pulsar-init-mqqv9                 0/1     Completed          0          4d21h
pulsar-pulsar-manager-76c5cfb97f-8xzcg   1/1     Running            0          4d21h
pulsar-recovery-0                        1/1     Running            2          4d21h
pulsar-toolset-0                         1/1     Running            0          4d21h
pulsar-zookeeper-0                       1/1     Running            0          4d21h
pulsar-zookeeper-1                       1/1     Running            0          4d21h
pulsar-zookeeper-2                       1/1     Running            0          4d21h

Error:

10:28:34.474 [pulsar-ordered-OrderedExecutor-1-0] INFO  org.apache.zookeeper.ZooKeeper - Initiating client connection, connectString=pulsar-zookeeper:2181 sessionTimeout=30000 watcher=org.apache.bookkeeper.zookeeper.ZooKeeperWatcherBase@198ff7fe
10:28:34.479 [pulsar-ordered-OrderedExecutor-1-0] INFO  org.apache.zookeeper.common.X509Util - Setting -D jdk.tls.rejectClientInitiatedRenegotiation=true to disable client-initiated TLS renegotiation
10:28:34.485 [pulsar-ordered-OrderedExecutor-1-0] INFO  org.apache.zookeeper.ClientCnxnSocket - jute.maxbuffer value is 10485760 Bytes
10:28:34.542 [pulsar-ordered-OrderedExecutor-1-0] INFO  org.apache.zookeeper.ClientCnxn - zookeeper.request.timeout value is 0. feature enabled=
10:28:54.559 [pulsar-ordered-OrderedExecutor-1-0-SendThread()] ERROR org.apache.zookeeper.client.StaticHostProvider - Unable to resolve address: pulsar-zookeeper:2181
java.net.UnknownHostException: pulsar-zookeeper: Temporary failure in name resolution
	at java.net.Inet4AddressImpl.lookupAllHostAddr(Native Method) ~[?:1.8.0_282]
	at java.net.InetAddress$2.lookupAllHostAddr(InetAddress.java:929) ~[?:1.8.0_282]
	at java.net.InetAddress.getAddressesFromNameService(InetAddress.java:1324) ~[?:1.8.0_282]
	at java.net.InetAddress.getAllByName0(InetAddress.java:1277) ~[?:1.8.0_282]
	at java.net.InetAddress.getAllByName(InetAddress.java:1193) ~[?:1.8.0_282]
	at java.net.InetAddress.getAllByName(InetAddress.java:1127) ~[?:1.8.0_282]
	at org.apache.zookeeper.client.StaticHostProvider$1.getAllByName(StaticHostProvider.java:92) ~[org.apache.pulsar-pulsar-zookeeper-2.7.2.jar:2.7.2]
	at org.apache.zookeeper.client.StaticHostProvider.resolve(StaticHostProvider.java:147) [org.apache.pulsar-pulsar-zookeeper-2.7.2.jar:2.7.2]
	at org.apache.zookeeper.client.StaticHostProvider.next(StaticHostProvider.java:375) [org.apache.pulsar-pulsar-zookeeper-2.7.2.jar:2.7.2]
	at org.apache.zookeeper.ClientCnxn$SendThread.run(ClientCnxn.java:1137) [org.apache.pulsar-pulsar-zookeeper-2.7.2.jar:2.7.2]
10:28:54.570 [pulsar-ordered-OrderedExecutor-1-0-SendThread(pulsar-zookeeper:2181)] WARN  org.apache.zookeeper.ClientCnxn - Session 0x0 for server pulsar-zookeeper:2181, unexpected error, closing socket connection and attempting reconnect
java.lang.IllegalArgumentException: Unable to canonicalize address pulsar-zookeeper:2181 because it's not resolvable

Do Anyone know about this Error? Please let me know how to Fix this Error?

This message was deleted.

Hi folks * * Description I'm trying to add JWT authentication and Authorization on a standalone apache pulsar, but seems it is not working for me. So basically right now the state is: Without token -- Client is able to produce and consume messages. To Reproduce Steps to reproduce 1. Create Tenant with admin roles pulsar-admin tenants create my-first-tenant --admin-roles testadmin 1.  Create Namespace under tenant pulsar-admin namespaces create my-first-tenant/my-first-namespace 1.  Create non-partitioned topic pulsar-admin topics create persistent://my-first-tenant/my-first-namespace/my-first-topic pulsar-admin namespaces grant-permission my-first-tenant/my-first-namespace --role testadmin --actions produce,consume 1.  Create secret key pulsar tokens create-secret-key --output /home/xxx/secret.key --base64 1.  Generate token using secret key pulsar tokens create --secret-key <file:///home/xxx/secret.key> --subject testadmin Changes made to the broker.conf • authenticationProviders=org.apache.pulsar.broker.authentication.AuthenticationProviderToken • authorizationProvider=org.apache.pulsar.broker.authorization.PulsarAuthorizationProvider • brokerClientTlsEnabled=true #false • brokerClientAuthenticationPlugin=org.apache.pulsar.client.impl.auth.AuthenticationToken # • brokerClientAuthenticationParameters={"token":"value of token"} • tokenSecretKey=file:///home/ankita/Acrolinx/apache-pulsar-2.9.1/tokens/secret.key • proxyRoles=testadmin • superUserRoles=testadmin Chnages made to proxy.conf • authenticationEnabled=true • authorizationEnabled=true • authenticationProviders=org.apache.pulsar.broker.authentication.AuthenticationProviderToken • tokenSecretKey=file:///path/to/secret.key • superUserRoles=testadmin • brokerClientAuthenticationPlugin=org.apache.pulsar.client.impl.auth.AuthenticationToken • brokerClientAuthenticationParameters={"token":"value of token"} • forwardAuthorizationCredentials=true Changes made to Standalone.conf • authenticationEnabled=true • authorizationEnabled=true • authenticationProviders=org.apache.pulsar.broker.authentication.AuthenticationProviderToken • tokenSecretKey=file:///path/to/secret.key • superUserRoles=testadmin • brokerClientAuthenticationPlugin=org.apache.pulsar.client.impl.auth.AuthenticationToken • brokerClientAuthenticationParameters={"token":"value of token"} Changes made to client.conf • authPlugin=org.apache.pulsar.client.impl.auth.AuthenticationToken • authParams=token:value of token Expected behaviour: I expect the client should not be able to produce or consume message without token Desktop -- Debian Additional Context I have attached python client files Any help is appreciated. Thanks in advance!

This message was deleted.

This message was deleted.

I have a multi-node cluster running on bare metal. There are 3 serves each with a node running broker, bookie, and zookeeper. I've tried initializing a cluster for the multi-nodes, but everytime it returns null and often with a concurrentCompletionException. Does anyone know how to resolve this and setup a multi-node bare metal cluster? I've read the Pulsar and suspecting it could be a zookeeper issue, but not sure.

This message was deleted.

This message was deleted.

This message was deleted.

This message was deleted.

Hello, I’m facing issue with using internal consumer. My use case: • partitioned topic • consumer: failover subscription type with event listener Pulsar broker notifies partition changes to consumer event listener at every partition re-assignment.

@Override
    public void becameActive(Consumer<?> consumer, int i) {

The problem is: I could not receive all messages from that partition via the given internal consumer of becameActive function.

Consumer<String> consumer = client.newConsumer(Schema.STRING)
                    .topic(TOPIC)
                    .subscriptionName("subscription-1")
                    .consumerEventListener(eventListener)
                    .subscriptionType(SubscriptionType.Failover)
                    .subscribe();

Does anyone has any ideas on this matter? I leave my test program here. You can try it with java

testcontainers

dependency

<dependency>
    <groupId>org.testcontainers</groupId>
    <artifactId>testcontainers</artifactId>
    <version>1.16.2</version>
    <scope>test</scope>
</dependency>
<dependency>
   <groupId>org.testcontainers</groupId>
   <artifactId>pulsar</artifactId>
   <version>1.16.2</version>
   <scope>test</scope>
</dependency>