https://pulsar.apache.org/ logo
Join Slack
Powered by
```I have a simple script attached. Both the test ...
# general
g
Copy code
I have a simple script attached. Both the test host and the pulsar cluster are in the same Azure resource group.

The producer can connect fine and send a message no problem. I can verify this with pulsar-admin:

pulsar-broker-0:/pulsar$ pulsar-admin topics examine-messages -i earliest <persistent://public/default/test-topic-1764699674>
Message ID: 8:0
Publish time: 1764699674223
Event time: 0
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 66 69 72 73 74 20 6d 65 73 73 61 67 65 20 66 6f |first message fo|
|00000010| 72 20 74 6f 70 69 63 20 74 65 73 74 2d 74 6f 70 |r topic test-top|
|00000020| 69 63 2d 31 37 36 34 36 39 39 36 37 34          |ic-1764699674   |
+--------+-------------------------------------------------+----------------+

Then the consumer can connect and create a subscription:

pulsar-broker-0:/pulsar$ pulsar-admin topics subscriptions <persistent://public/default/test-topic-1764699674>
test-subscription-1764699674

But it can't read because it times out.

Consumer log, note the error after the consumer is created:
============================================================
Apache Pulsar Basic Functionality Test
Started at: 2025-12-02T18:21:14.152135
============================================================
Topic: public/default/test-topic-1764699674
Subscription: test-subscription-1764699674
[     0ms] Created topic and subscription names
[    20ms] Created Pulsar client
2025-12-02 18:21:14.172 INFO  [139864562563968] ClientConnection:193 | [<none> -> <pulsar://pulsar-broker-1.pulsar-broker.pulsar-poc.svc.cluster.local:6650>] Create ClientConnection, timeout=10000
2025-12-02 18:21:14.172 INFO  [139864562563968] ConnectionPool:124 | Created connection for <pulsar://pulsar-broker-1.pulsar-broker.pulsar-poc.svc.cluster.local:6650>-<pulsar://pulsar-broker-1.pulsar-broker.pulsar-poc.svc.cluster.local:6650>-0
2025-12-02 18:21:14.183 INFO  [139864529450688] ClientConnection:410 | [10.20.3.78:56274 -> 10.20.4.29:6650] Connected to broker
2025-12-02 18:21:14.188 INFO  [139864529450688] HandlerBase:115 | [<persistent://public/default/test-topic-1764699674>, ] Getting connection from pool
2025-12-02 18:21:14.188 INFO  [139864529450688] BinaryProtoLookupService:85 | Lookup response for <persistent://public/default/test-topic-1764699674>, lookup-broker-url <pulsar://pulsar-broker-1.pulsar-broker.pulsar-poc.svc.cluster.local:6650>, from [10.20.3.78:56274 -> 10.20.4.29:6650]
2025-12-02 18:21:14.188 INFO  [139864529450688] ProducerImpl:147 | Creating producer for topic:<persistent://public/default/test-topic-1764699674>, producerName: on [10.20.3.78:56274 -> 10.20.4.29:6650]
2025-12-02 18:21:14.222 INFO  [139864529450688] ProducerImpl:220 | [<persistent://public/default/test-topic-1764699674>, ] Created producer on broker [10.20.3.78:56274 -> 10.20.4.29:6650]
2025-12-02 18:21:14.222 INFO  [139864529450688] HandlerBase:138 | Finished connecting to broker after 34 ms
[    50ms] Created producer
[    28ms] Sent message: 'first message for topic test-topic-1764699674'
2025-12-02 18:21:14.250 INFO  [139864562563968] Client:86 | Subscribing on Topic :public/default/test-topic-1764699674
2025-12-02 18:21:14.251 INFO  [139864529450688] HandlerBase:115 | [<persistent://public/default/test-topic-1764699674>, test-subscription-1764699674, 0] Getting connection from pool
2025-12-02 18:21:14.252 INFO  [139864529450688] BinaryProtoLookupService:85 | Lookup response for <persistent://public/default/test-topic-1764699674>, lookup-broker-url <pulsar://pulsar-broker-1.pulsar-broker.pulsar-poc.svc.cluster.local:6650>, from [10.20.3.78:56274 -> 10.20.4.29:6650]
2025-12-02 18:21:14.260 INFO  [139864529450688] ConsumerImpl:311 | [<persistent://public/default/test-topic-1764699674>, test-subscription-1764699674, 0] Created consumer on broker [10.20.3.78:56274 -> 10.20.4.29:6650]
2025-12-02 18:21:14.260 INFO  [139864529450688] HandlerBase:138 | Finished connecting to broker after 8 ms
[    10ms] Created consumer
Copy code
ERROR: Timeout while trying to receive message: Pulsar error: TimeOut 
2025-12-02 18:21:19.260 INFO  [139864562563968] ConsumerImpl:1328 | [<persistent://public/default/test-topic-1764699674>, test-subscription-1764699674, 0] Closing consumer for topic <persistent://public/default/test-topic-1764699674>
2025-12-02 18:21:19.261 INFO  [139864529450688] ConsumerImpl:1312 | [<persistent://public/default/test-topic-1764699674>, test-subscription-1764699674, 0] Closed consumer 0
Consumer closed
2025-12-02 18:21:19.261 INFO  [139864562563968] ProducerImpl:800 | [<persistent://public/default/test-topic-1764699674>, pulsar-1-7] Closing producer for topic <persistent://public/default/test-topic-1764699674>
2025-12-02 18:21:19.261 INFO  [139864529450688] ProducerImpl:764 | [<persistent://public/default/test-topic-1764699674>, pulsar-1-7] Closed producer 0
Producer closed
2025-12-02 18:21:19.261 INFO  [139864562563968] ClientImpl:665 | Closing Pulsar client with 0 producers and 0 consumers
2025-12-02 18:21:19.261 INFO  [139864504272576] ClientConnection:1336 | [10.20.3.78:56274 -> 10.20.4.29:6650] Connection disconnected (refCnt: 1)
2025-12-02 18:21:19.261 INFO  [139864504272576] ClientConnection:282 | [10.20.3.78:56274 -> 10.20.4.29:6650] Destroyed connection to <pulsar://pulsar-broker-1.pulsar-broker.pulsar-poc.svc.cluster.local:6650>-0
Client closed
2025-12-02 18:21:19.261 INFO  [139864562563968] ProducerImpl:757 | Producer - [<persistent://public/default/test-topic-1764699674>, pulsar-1-7] , [batching  = off]

Relevant broker logs:
2025-12-02T18:21:14,186+0000 [pulsar-io-3-25] INFO  org.apache.pulsar.broker.service.ServerCnx - [/10.20.3.78:56274] connected with clientVersion=Pulsar-CPP-v3.7.2, clientProtocolVersion=20, proxyVersion=null
2025-12-02T18:21:14,193+0000 [ForkJoinPool.commonPool-worker-1651] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - Opening managed ledger public/default/persistent/test-topic-1764699674
2025-12-02T18:21:14,197+0000 [bookkeeper-ml-scheduler-OrderedScheduler-11-0] INFO  org.apache.bookkeeper.mledger.impl.MetaStoreImpl - Creating '/managed-ledgers/public/default/persistent/test-topic-1764699674'
2025-12-02T18:21:14,205+0000 [metadata-store-10-1] INFO  org.apache.bookkeeper.client.BookieWatcherImpl - New ensemble: [pulsar-bookie-1.pulsar-bookie.pulsar-poc.svc.cluster.local:3181, pulsar-bookie-0.pulsar-bookie.pulsar-poc.svc.cluster.local:3181] is not adhering to Placement Policy. quarantinedBookies: []
2025-12-02T18:21:14,210+0000 [ForkJoinPool.commonPool-worker-1-EventThread] INFO  org.apache.bookkeeper.client.LedgerCreateOp - Ensemble: [pulsar-bookie-1.pulsar-bookie.pulsar-poc.svc.cluster.local:3181, pulsar-bookie-0.pulsar-bookie.pulsar-poc.svc.cluster.local:3181] for ledger: 8
2025-12-02T18:21:14,210+0000 [BookKeeperClientWorker-OrderedExecutor-11-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [public/default/persistent/test-topic-1764699674] Created ledger 8 after closed null
2025-12-02T18:21:14,217+0000 [bookkeeper-ml-scheduler-OrderedScheduler-11-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl - [public/default/persistent/test-topic-1764699674] Successfully initialize managed ledger
2025-12-02T18:21:14,218+0000 [broker-topic-workers-OrderedExecutor-0-0] INFO  org.apache.pulsar.broker.service.BrokerService - Created topic <persistent://public/default/test-topic-1764699674> - dedup is disabled (latency: 28 ms)
2025-12-02T18:21:14,223+0000 [broker-topic-workers-OrderedExecutor-4-0] INFO  org.apache.pulsar.broker.service.ServerCnx - [/10.20.3.78:56274] Created new producer: Producer{topic=PersistentTopic{topic=<persistent://public/default/test-topic-1764699674>}, client=[id: 0x244e0302, L:/10.20.4.29:6650 - R:/10.20.3.78:56274] [SR:10.20.3.78, state:Connected], producerName=pulsar-1-7, producerId=0}, role: null
2025-12-02T18:21:14,253+0000 [pulsar-io-3-25] INFO  org.apache.pulsar.broker.service.ServerCnx - [[id: 0x244e0302, L:/10.20.4.29:6650 - R:/10.20.3.78:56274] [SR:10.20.3.78, state:Connected]] Subscribing on topic <persistent://public/default/test-topic-1764699674> / test-subscription-1764699674. consumerId: 0, role: null
2025-12-02T18:21:14,253+0000 [ForkJoinPool.commonPool-worker-1651] INFO  org.apache.bookkeeper.mledger.impl.ManagedCursorImpl - [public/default/persistent/test-topic-1764699674] Cursor test-subscription-1764699674 recovered to position 8:0
2025-12-02T18:21:14,260+0000 [bookkeeper-ml-scheduler-OrderedScheduler-11-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl - [public/default/persistent/test-topic-1764699674] Opened new cursor: ManagedCursorImpl{ledger=public/default/persistent/test-topic-1764699674, name=test-subscription-1764699674, ackPos=8:0, readPos=8:1}
2025-12-02T18:21:14,260+0000 [bookkeeper-ml-scheduler-OrderedScheduler-11-0] INFO  org.apache.bookkeeper.mledger.impl.ManagedCursorImpl - [public/default/persistent/test-topic-1764699674-test-subscription-1764699674] Rewind from 8:1 to 8:1
2025-12-02T18:21:14,260+0000 [bookkeeper-ml-scheduler-OrderedScheduler-11-0] INFO  org.apache.pulsar.broker.service.persistent.PersistentSubscription - backlog for <persistent://public/default/test-topic-1764699674> - 0
2025-12-02T18:21:14,260+0000 [bookkeeper-ml-scheduler-OrderedScheduler-11-0] INFO  org.apache.pulsar.broker.service.ServerCnx - [/10.20.3.78:56274] Created subscription on topic <persistent://public/default/test-topic-1764699674> / test-subscription-1764699674
2025-12-02T18:21:19,261+0000 [pulsar-io-3-25] INFO  org.apache.pulsar.broker.service.ServerCnx - [/10.20.3.78:56274] Closing consumer: consumerId=0
2025-12-02T18:21:19,261+0000 [pulsar-io-3-25] INFO  org.apache.pulsar.broker.service.AbstractDispatcherSingleActiveConsumer - Removing consumer Consumer{subscription=PersistentSubscription{topic=<persistent://public/default/test-topic-1764699674>, name=test-subscription-1764699674}, consumerId=0, consumerName=9ba66a395e, address=[id: 0x244e0302, L:/10.20.4.29:6650 - R:/10.20.3.78:56274] [SR:10.20.3.78, state:Connected]}
2025-12-02T18:21:19,261+0000 [pulsar-io-3-25] INFO  org.apache.pulsar.broker.service.ServerCnx - [/10.20.3.78:56274] Closed consumer, consumerId=0
2025-12-02T18:21:19,262+0000 [pulsar-io-3-25] INFO  org.apache.pulsar.broker.service.ServerCnx - [PersistentTopic{topic=<persistent://public/default/test-topic-1764699674}][pulsar-1-7>] Closing producer on cnx /10.20.3.78:56274. producerId=0
2025-12-02T18:21:19,262+0000 [pulsar-io-3-25] INFO  org.apache.pulsar.broker.service.ServerCnx - [PersistentTopic{topic=<persistent://public/default/test-topic-1764699674}][pulsar-1-7>] Closed producer on cnx /10.20.3.78:56274. producerId=0
2025-12-02T18:21:19,262+0000 [pulsar-io-3-25] INFO  org.apache.pulsar.broker.service.ServerCnx - Closed connection from /10.20.3.78:56274
2025-12-02T18:22:56,867+0000 [pulsar-io-3-27] INFO  org.apache.pulsar.broker.service.ServerCnx - [/10.20.3.78:59534] connected with clientVersion=Pulsar-CPP-v3.7.2, clientProtocolVersion=20, proxyVersion=null
2025-12-02T18:23:02,569+0000 [pulsar-io-3-27] INFO  org.apache.pulsar.broker.service.ServerCnx - Closed connection from /10.20.3.78:59534

Any assistance in diagnosing this would be appreciated. Otherwise, this is a full stop for us in terms of our evaluation of migrating from Kafka to Pulsar.

Best,

Glenn
pulsar_basic_test.py
d
Given the way Pulsar subscriptions work, you need to create the subscription (step 5 in your script) before producing messages (step 3).
g
That's interesting. So, 1. Create producer on topic (so that the topic exists) 2. Create consumer on topic, sub (so that sub exists) 3. Create message 4. Read message. Does that sound right?
d
That should work.
By default, Pulsar doesn’t retain messages on a topic unless there is an active subscription. When a subscription is created, it essentially means “start reading messages from this point in time forward”, so previously published messages on the topic are ignored.
✅ 1
👍 1
g
understood, thanks for the explanation!
Out of curiosity, if it is a shared subscription, do all consumers need to attach first or just one?
d
Just one. Creation of a subscription is all that is required. Consumers can join and leave the group after the fact.
g
Awesome, you've been a huge help, much appreciated!
🙏 1
Mucho gracias!!!!
Copy code
============================================================
Apache Pulsar Basic Functionality Test
Started at: 2025-12-02T20:44:48.366132
============================================================
Topic: public/default/test-topic-1764708288
Subscription: test-subscription-1764708288
[     0ms] Created topic and subscription names
[    19ms] Created Pulsar client
2025-12-02 20:44:48.386 INFO  [140324107926400] ClientConnection:193 | [<none> -> <pulsar://pulsar-broker-1.pulsar-broker.pulsar-poc.svc.cluster.local:6650>] Create ClientConnection, timeout=10000
2025-12-02 20:44:48.386 INFO  [140324107926400] ConnectionPool:124 | Created connection for <pulsar://pulsar-broker-1.pulsar-broker.pulsar-poc.svc.cluster.local:6650>-<pulsar://pulsar-broker-1.pulsar-broker.pulsar-poc.svc.cluster.local:6650>-0
2025-12-02 20:44:48.402 INFO  [140324074813120] ClientConnection:410 | [10.20.3.78:49288 -> 10.20.4.29:6650] Connected to broker
2025-12-02 20:44:48.408 INFO  [140324074813120] HandlerBase:115 | [<persistent://public/default/test-topic-1764708288>, ] Getting connection from pool
2025-12-02 20:44:48.422 INFO  [140324074813120] ClientConnection:193 | [<none> -> <pulsar://pulsar-broker-2.pulsar-broker.pulsar-poc.svc.cluster.local:6650>] Create ClientConnection, timeout=10000
2025-12-02 20:44:48.422 INFO  [140324074813120] ConnectionPool:124 | Created connection for <pulsar://pulsar-broker-2.pulsar-broker.pulsar-poc.svc.cluster.local:6650>-<pulsar://pulsar-broker-2.pulsar-broker.pulsar-poc.svc.cluster.local:6650>-0
2025-12-02 20:44:48.429 INFO  [140324074813120] ClientConnection:410 | [10.20.3.78:39436 -> 10.20.6.144:6650] Connected to broker
2025-12-02 20:44:48.473 INFO  [140324074813120] BinaryProtoLookupService:85 | Lookup response for <persistent://public/default/test-topic-1764708288>, lookup-broker-url <pulsar://pulsar-broker-1.pulsar-broker.pulsar-poc.svc.cluster.local:6650>, from [10.20.3.78:49288 -> 10.20.4.29:6650]
2025-12-02 20:44:48.473 INFO  [140324074813120] ProducerImpl:147 | Creating producer for topic:<persistent://public/default/test-topic-1764708288>, producerName: on [10.20.3.78:49288 -> 10.20.4.29:6650]
2025-12-02 20:44:48.522 INFO  [140324074813120] ProducerImpl:220 | [<persistent://public/default/test-topic-1764708288>, ] Created producer on broker [10.20.3.78:49288 -> 10.20.4.29:6650]
2025-12-02 20:44:48.522 INFO  [140324074813120] HandlerBase:138 | Finished connecting to broker after 114 ms
[   138ms] Created producer
2025-12-02 20:44:48.523 INFO  [140324107926400] Client:86 | Subscribing on Topic :public/default/test-topic-1764708288
2025-12-02 20:44:48.523 INFO  [140324074813120] HandlerBase:115 | [<persistent://public/default/test-topic-1764708288>, test-subscription-1764708288, 0] Getting connection from pool
2025-12-02 20:44:48.524 INFO  [140324074813120] BinaryProtoLookupService:85 | Lookup response for <persistent://public/default/test-topic-1764708288>, lookup-broker-url <pulsar://pulsar-broker-1.pulsar-broker.pulsar-poc.svc.cluster.local:6650>, from [10.20.3.78:49288 -> 10.20.4.29:6650]
2025-12-02 20:44:48.529 INFO  [140324074813120] ConsumerImpl:311 | [<persistent://public/default/test-topic-1764708288>, test-subscription-1764708288, 0] Created consumer on broker [10.20.3.78:49288 -> 10.20.4.29:6650]
2025-12-02 20:44:48.529 INFO  [140324074813120] HandlerBase:138 | Finished connecting to broker after 5 ms
[     6ms] Created consumer
[    19ms] Sent message: 'first message for topic test-topic-1764708288'
[     1ms] Received message: 'first message for topic test-topic-1764708288'
[     0ms] ✓ Message verified and acknowledged

============================================================
Test Result: SUCCESS
============================================================
2025-12-02 20:44:48.549 INFO  [140324107926400] ConsumerImpl:1328 | [<persistent://public/default/test-topic-1764708288>, test-subscription-1764708288, 0] Closing consumer for topic <persistent://public/default/test-topic-1764708288>
2025-12-02 20:44:48.572 INFO  [140324074813120] ConsumerImpl:1312 | [<persistent://public/default/test-topic-1764708288>, test-subscription-1764708288, 0] Closed consumer 0
Consumer closed
2025-12-02 20:44:48.573 INFO  [140324107926400] ProducerImpl:800 | [<persistent://public/default/test-topic-1764708288>, pulsar-1-8] Closing producer for topic <persistent://public/default/test-topic-1764708288>
2025-12-02 20:44:48.573 INFO  [140324074813120] ProducerImpl:764 | [<persistent://public/default/test-topic-1764708288>, pulsar-1-8] Closed producer 0
Producer closed
2025-12-02 20:44:48.573 INFO  [140324107926400] ClientImpl:665 | Closing Pulsar client with 0 producers and 0 consumers
2025-12-02 20:44:48.573 INFO  [140324049635008] ClientConnection:1336 | [10.20.3.78:49288 -> 10.20.4.29:6650] Connection disconnected (refCnt: 1)
2025-12-02 20:44:48.573 INFO  [140324049635008] ClientConnection:1336 | [10.20.3.78:39436 -> 10.20.6.144:6650] Connection disconnected (refCnt: 2)
2025-12-02 20:44:48.573 INFO  [140324049635008] ClientConnection:282 | [10.20.3.78:49288 -> 10.20.4.29:6650] Destroyed connection to <pulsar://pulsar-broker-1.pulsar-broker.pulsar-poc.svc.cluster.local:6650>-0
2025-12-02 20:44:48.573 INFO  [140324074813120] ClientConnection:282 | [10.20.3.78:39436 -> 10.20.6.144:6650] Destroyed connection to <pulsar://pulsar-broker-2.pulsar-broker.pulsar-poc.svc.cluster.local:6650>-0
Client closed
2025-12-02 20:44:48.574 INFO  [140324107926400] ProducerImpl:757 | Producer - [<persistent://public/default/test-topic-1764708288>, pulsar-1-8] , [batching  = off]
👏 2
Open in Slack
PreviousNext
Powered by