You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "namest504 (via GitHub)" <gi...@apache.org> on 2026/10/21 02:45:54 UTC
[PR] [fix][client] Close the producer when its encryption key cannot be loaded [pulsar]
namest504 opened a new pull request, #27024:
URL: https://github.com/apache/pulsar/pull/27024
Fixes #24280
### Motivation
When the `CryptoKeyReader` of a producer cannot provide the public key, `create()` fails with `CryptoException`, as `SimpleProducerConsumerTest#testEncryptionFailure` requires. But the producer was left behind: the key is loaded by a task on the event loop, and on failure that task only completed `producerCreatedFuture` exceptionally. It did not close the producer or remove it from the client, unlike the other creation failures. Meanwhile `grabCnx()` had already started, so the connection handler registered the producer on the broker, and because the creation future was already completed, every later connection failure was treated as a reconnection. An application that retries `create()` accumulates producers on the broker.
### Modifications
- `ProducerImpl`: when adding the public key cipher fails during creation, close the producer after failing the creation future. The close moves it to `Closed`, cancels its tasks and removes it from the client; a connection opened afterwards sees the closed state and does not register the producer.
- `PulsarClientImpl`: remove a producer from the client when its creation future fails. The key task runs on the event loop and can fail before `createProducerAsync` has added the producer to the client, in which case the close above has nothing to remove yet.
- `ProducerEncryptionKeyFailureTest` (new, broker test): a producer with an encryption key and a key reader returning `null`. `create()` fails, the client holds no producer, and the broker shows no publisher on the topic for three seconds. The test fails before the change on both counts.
### Verifying this change
- [x] Make sure that the change passes the CI checks.
`ProducerEncryptionKeyFailureTest` and `SimpleProducerConsumerTest#testEncryptionFailure` pass locally.
### Does this pull request potentially affect one of the following parts:
- [ ] Dependencies
- [ ] The public API
- [ ] The schema
- [ ] The default values of configurations
- [ ] The threading model
- [ ] The binary protocol
- [ ] The REST endpoints
- [ ] The admin CLI options
- [ ] The metrics
- [ ] Anything that affects deployment
### Documentation
- [x] `doc-not-needed`
--
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
Re: [PR] [fix][client] Close the producer when its encryption key cannot be loaded [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui merged PR #27024:
URL: https://github.com/apache/pulsar/pull/27024
--
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
Re: [PR] [fix][client] Close the producer when its encryption key cannot be loaded [pulsar]
Posted by "namest504 (via GitHub)" <gi...@apache.org>.
namest504 commented on code in PR #27024:
URL: https://github.com/apache/pulsar/pull/27024#discussion_r4343486401
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -3168,6 +3153,40 @@ void setClientCnx(ClientCnx clientCnx) {
this.connectionHandler.setClientCnx(clientCnx);
}
+ /**
+ * Loads the public key cipher now and refreshes it at a fixed interval. Called once the producer is fully
+ * initialized: a key that cannot be loaded fails the creation and closes the producer, which uses the counters,
+ * the stats and the connection handler. {@link #closeAsync()} is synchronized, so the task cannot close the
+ * producer before the handle of the task is stored here and the close can cancel it; when the task has run by
+ * then, from the scheduling thread for instance, the handle is cancelled here instead.
+ */
+ private synchronized void scheduleKeyGeneratorTask() {
+ keyGeneratorTask = client.eventLoopGroup().scheduleWithFixedDelay(
+ catchingAndLoggingThrowables(this::addPublicKeyCipher), 0L, 4L, TimeUnit.HOURS);
+ if (getState() == State.Closing || getState() == State.Closed) {
+ keyGeneratorTask.cancel(false);
Review Comment:
Thanks. Two changes in e8fd3a5dba. The task is scheduled before `grabCnx()` again, right after the connection handler is initialized, so a connection failure that ends the creation finds the handle in `closeProducerTasks()`; this also closes the window where `closeProducerTasks()` runs before `setState(Failed)` on the connection thread while the constructor is still between storing the handle and checking the state. And the check after scheduling now covers every terminal state (`Closing`, `Closed`, `Failed`, `Terminated`, `ProducerFenced`) through a small `ProducerBase#isTerminalState()`, which `createProducerAsync` uses as well, so a producer that failed synchronously before it was added to the client is not kept either.
`testKeyTaskIsCancelledWhenTheConnectionFailsTheCreation` is your scenario: an immediately failed connection future with an `AuthorizationException`; the creation fails, the state is `Failed`, the producer is removed from the client and the handle is cancelled. `testProducerEndedBeforeItIsRegisteredIsNotKept` now also covers the `Failed` case on the client side.
--
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
Re: [PR] [fix][client] Close the producer when its encryption key cannot be loaded [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #27024:
URL: https://github.com/apache/pulsar/pull/27024#discussion_r4343230948
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -3168,6 +3153,40 @@ void setClientCnx(ClientCnx clientCnx) {
this.connectionHandler.setClientCnx(clientCnx);
}
+ /**
+ * Loads the public key cipher now and refreshes it at a fixed interval. Called once the producer is fully
+ * initialized: a key that cannot be loaded fails the creation and closes the producer, which uses the counters,
+ * the stats and the connection handler. {@link #closeAsync()} is synchronized, so the task cannot close the
+ * producer before the handle of the task is stored here and the close can cancel it; when the task has run by
+ * then, from the scheduling thread for instance, the handle is cancelled here instead.
+ */
+ private synchronized void scheduleKeyGeneratorTask() {
+ keyGeneratorTask = client.eventLoopGroup().scheduleWithFixedDelay(
+ catchingAndLoggingThrowables(this::addPublicKeyCipher), 0L, 4L, TimeUnit.HOURS);
+ if (getState() == State.Closing || getState() == State.Closed) {
+ keyGeneratorTask.cancel(false);
Review Comment:
**[P2] Do not start key refresh after terminal connection cleanup.** The constructor now calls `grabCnx()` before scheduling this task. If initial lookup/connection acquisition fails non-retriably first—for example, with an authorization failure—`connectionFailed()` runs `closeProducerTasks()` while `keyGeneratorTask` is still null, then enters `Failed`. This subsequently installs a recurring task that the `Closing`/`Closed`-only guard does not cancel. It retains the failed producer and can continue key-reader/crypto work every four hours; retries accumulate these tasks on a long-lived client.
A focused test supplied an immediately failed connection future, confirmed exceptional creation, `Failed` state and client cleanup, then failed because the scheduled handle's `cancel(false)` was never invoked. Previously that handle was assigned before `grabCnx()`.
Please schedule the task after initialization but before connection startup, or coordinate scheduling atomically with all terminal cleanup paths, including `Failed`, `Terminated` and `ProducerFenced`. The synchronization here protects against `closeAsync()`, but not `connectionFailed()` cleanup. Please add deterministic coverage for this ordering.
--
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
Re: [PR] [fix][client] Close the producer when its encryption key cannot be loaded [pulsar]
Posted by "namest504 (via GitHub)" <gi...@apache.org>.
namest504 commented on code in PR #27024:
URL: https://github.com/apache/pulsar/pull/27024#discussion_r4333176395
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -281,6 +281,14 @@ public ProducerImpl(PulsarClientImpl client, String topic, ProducerConfiguration
String.format("The producer %s of the topic %s "
+ "adds the public key cipher was failed",
producerName, topic)));
+ // The creation has failed, so close the producer as the other creation failures do.
+ // Otherwise it stays in the client and keeps the connection it is about to open, which
+ // registers it on the broker and then reconnects it as if it had been created.
+ closeAsync().exceptionally(closeEx -> {
Review Comment:
You're right, the task could run before the counters and the connection handler existed, and the close then threw before it could cancel anything. The key task is now scheduled as the last step of the constructor, from a synchronized method that stores the handle and cancels it if the task has already closed the producer by then; `closeAsync()` is synchronized too, so a task on another thread cannot close the producer before the handle is stored. `testKeyThatCannotBeLoadedClosesTheProducer` covers both timings: the task run from within `scheduleWithFixedDelay` (worst case, same thread) and after the constructor returns; in both the creation fails with `CryptoException`, the state is `Closed`, the handle is cancelled and the producer is removed from the client.
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java:
##########
@@ -1013,6 +1013,12 @@ private <T> CompletableFuture<Producer<T>> createProducerAsync(String topic,
Optional.empty());
}
producers.add(producer);
+ // The creation can fail before this line runs, from the event loop, and then the failure has already
+ // tried to remove the producer. Keep a producer that was never created out of the client.
+ producerCreatedFuture.exceptionally(ex -> {
+ cleanupProducer(producer);
+ return null;
Review Comment:
Good catch on the cancellation. The hook on the creation future is gone; instead `createProducerAsync` checks the producer's state right after adding it and removes it only when it is already closing or closed, which is the case this change is about (the key task closing the producer from another thread before the add). Cancelling the returned future therefore behaves as on master: the producer stays owned by the client and is closed with it. `testProducerClosedBeforeItIsRegisteredIsNotKept` covers the check.
--
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
Re: [PR] [fix][client] Close the producer when its encryption key cannot be loaded [pulsar]
Posted by "namest504 (via GitHub)" <gi...@apache.org>.
namest504 commented on code in PR #27024:
URL: https://github.com/apache/pulsar/pull/27024#discussion_r4356451770
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java:
##########
@@ -1013,6 +1013,12 @@ private <T> CompletableFuture<Producer<T>> createProducerAsync(String topic,
Optional.empty());
}
producers.add(producer);
+ // The creation can fail before this line runs, from another thread or from a connection that fails at
+ // once, in which case the producer has already closed or failed and removed itself from the client. Do
+ // not keep it then.
+ if (producer.isTerminalState()) {
+ cleanupProducer(producer);
+ }
Review Comment:
Follow-up: #27061.
--
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
Re: [PR] [fix][client] Close the producer when its encryption key cannot be loaded [pulsar]
Posted by "namest504 (via GitHub)" <gi...@apache.org>.
namest504 commented on code in PR #27024:
URL: https://github.com/apache/pulsar/pull/27024#discussion_r4333176599
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java:
##########
@@ -1013,6 +1013,12 @@ private <T> CompletableFuture<Producer<T>> createProducerAsync(String topic,
Optional.empty());
}
producers.add(producer);
+ // The creation can fail before this line runs, from the event loop, and then the failure has already
+ // tried to remove the producer. Keep a producer that was never created out of the client.
+ producerCreatedFuture.exceptionally(ex -> {
+ cleanupProducer(producer);
+ return null;
Review Comment:
Thanks. The hook on the creation future is gone; instead `createProducerAsync` checks the producer's state right after adding it and removes it only when it is already closing or closed, which is the case this change is about (the key task closing the producer from another thread before the add). Cancelling the returned future therefore behaves as on master: the producer stays owned by the client and is closed with it. `testProducerClosedBeforeItIsRegisteredIsNotKept` covers the check.
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java:
##########
@@ -1013,6 +1013,12 @@ private <T> CompletableFuture<Producer<T>> createProducerAsync(String topic,
Optional.empty());
}
producers.add(producer);
+ // The creation can fail before this line runs, from the event loop, and then the failure has already
+ // tried to remove the producer. Keep a producer that was never created out of the client.
+ producerCreatedFuture.exceptionally(ex -> {
+ cleanupProducer(producer);
+ return null;
Review Comment:
Added `testCancelledCreationLeavesTheProducerOwnedByTheClient` for the cancellation/shutdown side, for a non-partitioned and a two-partition topic: the creation future is cancelled before the metadata is known, the metadata then completes, and the producer comes up anyway; the client still counts it, and closing the client closes it. With the earlier hook on the creation future the test fails on the ownership check.
--
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
Re: [PR] [fix][client] Close the producer when its encryption key cannot be loaded [pulsar]
Posted by "namest504 (via GitHub)" <gi...@apache.org>.
namest504 commented on PR #27024:
URL: https://github.com/apache/pulsar/pull/27024#issuecomment-6331747380
Thanks. Merged master in f170bf5564; the only conflict was an import in `ProducerImplTest`. `ProducerImplTest` (now 52 tests with the ones from #27026), `PulsarClientImplTest`, `ProducerEncryptionKeyFailureTest` and `SimpleProducerConsumerTest#testEncryptionFailure` pass on the merged branch. I will open a follow-up for the partitioned producer timer after this is merged.
--
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
Re: [PR] [fix][client] Close the producer when its encryption key cannot be loaded [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #27024:
URL: https://github.com/apache/pulsar/pull/27024#discussion_r4333069402
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -281,6 +281,14 @@ public ProducerImpl(PulsarClientImpl client, String topic, ProducerConfiguration
String.format("The producer %s of the topic %s "
+ "adds the public key cipher was failed",
producerName, topic)));
+ // The creation has failed, so close the producer as the other creation failures do.
+ // Otherwise it stays in the client and keeps the connection it is about to open, which
+ // registers it on the broker and then reconnects it as if it had been created.
+ closeAsync().exceptionally(closeEx -> {
Review Comment:
**[P2] Defer teardown until producer initialization is complete.** The key task has zero initial delay and is scheduled before stats, counters and the connection handler are initialized. If it fails while the constructor is still running, this call sets the state to `Closing` and then throws synchronously because `producersClosedCounter` is null; the attached `exceptionally` cannot catch an exception thrown before a future is returned. Task cancellation and terminal cleanup are skipped, and subsequent close calls return immediately because the state is already `Closing`.
I forced this valid ordering through the batcher-builder hook in a review-only test, without modifying private state: it logged that NPE and failed with `Closed` expected but `Closing` observed. Please order key-task startup/cleanup after initialization, including availability of the scheduled handle for cancellation, and cover this interleaving deterministically.
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java:
##########
@@ -1013,6 +1013,12 @@ private <T> CompletableFuture<Producer<T>> createProducerAsync(String topic,
Optional.empty());
}
producers.add(producer);
+ // The creation can fail before this line runs, from the event loop, and then the failure has already
+ // tried to remove the producer. Keep a producer that was never created out of the client.
+ producerCreatedFuture.exceptionally(ex -> {
+ cleanupProducer(producer);
+ return null;
Review Comment:
**[P2] Retain shutdown ownership when creation is cancelled.** The ordinary `createAsync()` path returns this same creation future, so cancelling it also invokes this callback. Cancellation does not stop the underlying producer creation: it can still reach `Ready`, while `cleanupProducer()` merely removes it from the client set without closing it.
Before this change, the cancelled producer remained registered for `client.closeAsync()` to close. Now shutdown skips its producer/schema/task teardown; with externally shared connection pools and timers, broker registrations and scheduled work can survive client closure. Please either retain ownership for cancelled creation or explicitly close the producer before removal, with coverage for cancellation before metadata/connection completion, including partitioned producers.
--
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
Re: [PR] [fix][client] Close the producer when its encryption key cannot be loaded [pulsar]
Posted by "codelipenghui (via GitHub)" <gi...@apache.org>.
codelipenghui commented on code in PR #27024:
URL: https://github.com/apache/pulsar/pull/27024#discussion_r4343805977
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java:
##########
@@ -1013,6 +1013,12 @@ private <T> CompletableFuture<Producer<T>> createProducerAsync(String topic,
Optional.empty());
}
producers.add(producer);
+ // The creation can fail before this line runs, from another thread or from a connection that fails at
+ // once, in which case the producer has already closed or failed and removed itself from the client. Do
+ // not keep it then.
+ if (producer.isTerminalState()) {
+ cleanupProducer(producer);
+ }
Review Comment:
**[P3, pre-existing, non-blocking]** A partitioned producer that ends inside its constructor is correctly dropped here, but `PartitionedProducerImpl` (not touched by this PR) still keeps a timer for it.
If the first partition's creation fails before `start()` attaches its handler, the producer closes inside `start()`. A fast-failing key reader on the event loop can cause this, and so can a synchronous non-retriable connection failure. The constructor then schedules `partitionsAutoUpdateTimeout` anyway (`PartitionedProducerImpl.java:125–130`). The timer task's `finally` block reschedules it every `autoUpdatePartitionsIntervalSeconds`, even when the state is not `Ready` (lines 505–510). So each failed `create()` on a partitioned topic can leave behind a recurring task that keeps the closed producer and its children reachable.
A review-only deterministic test reproduced it: 1 task scheduled by the constructor, 2 after one run, none cancelled.
Fine as a follow-up PR: reschedule only while `!isTerminalState()`, and skip scheduling when `start()` has already ended the producer.
--
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