[pulsar-broker]Fix: client-producer can't connect due to failed producer-future on cnx
@@ -100,10 +100,8 @@ import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.BacklogQuota;
import org.apache.pulsar.common.policies.data.ConsumerStats;
-import org.apache.pulsar.common.sasl.SaslConstants;
import org.apache.pulsar.common.schema.SchemaData;
import org.apache.pulsar.common.schema.SchemaInfoUtil;
-import org.apache.pulsar.common.schema.SchemaType;
import org.apache.pulsar.common.schema.SchemaVersion;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.collections.ConcurrentLongHashMap;
@@ -851,8 +849,13 @@ public class ServerCnx extends PulsarHandler {
// until the previous producer creation
// request
// either complete or fails.
- ServerError error = !existingProducerFuture.isDone() ? ServerError.ServiceNotReady
- : getErrorCode(existingProducerFuture);
+ ServerError error = null;
+ if(!existingProducerFuture.isDone()) {
+ error = ServerError.ServiceNotReady;
+ }else {
+ error = getErrorCode(existingProducerFuture);
+ producers.remove(producerId, producerFuture);
+ }
log.warn("[{}][{}] Producer with id {} is already present on the connection", remoteAddress,
producerId, topicName);
ctx.writeAndFlush(Commands.newError(requestId, error,
@@ -899,6 +902,8 @@ public class ServerCnx extends PulsarHandler {
schemaVersionFuture = topic.addSchema(schema);
} else {
schemaVersionFuture = topic.hasSchema().thenCompose((hasSchema) -> {
+ log.info("[{}]-{} {} configured with schema {}", remoteAddress, producerId,
+ topicName, hasSchema);
CompletableFuture<SchemaVersion> result = new CompletableFuture<>();
if (hasSchema && schemaValidationEnforced) {
result.completeExceptionally(new IncompatibleSchemaException(
