Hi Everyone, In a k8S cluster , Out of 200 topic, If couple of topic is having backlog quota exceeded error and continuously we are pushing messages on that topic and all other topics aswell , Pulsar Client is taking almost 30sec to respond to create producer (due to this opertaiontimeout parameter 30 sec) for backlogquotaexception. Due to this behavior, For other topics throughput is getting dropped . Could anyone help me out , Is there any possibility to twick any pulsar parameter to handle this scenario?

Client creation code

client = PulsarClient.builder()
					.serviceUrl("<pulsar+ssl://10.5545767:90>:31051")
					.enableTls(true)
					//					.tlsProtocols(protocol)
					//					.maxNumberOfRejectedRequestPerConnection(50)
					//					.maxLookupRequests(100000)
					//					.maxLookupRedirects(50)
					//					.keepAliveInterval(100, TimeUnit.SECONDS)
					//					.connectionTimeout(100, TimeUnit.SECONDS)
					//.operationTimeout(90,TimeUnit.SECONDS)
					//					.maxBackoffInterval(100,TimeUnit.SECONDS)
					//					
					//					.startingBackoffInterval(300,TimeUnit.MILLISECONDS)
					.enableTlsHostnameVerification(false)
					.allowTlsInsecureConnection(false)
					.tlsTrustCertsFilePath("C:\\PI\\poc\\optic-test-new\\src\\main\\resources\\issue_ca.crt")
					.authentication("org.apache.pulsar.client.impl.auth.AuthenticationTls",authParams)
					//.ioThreads(2)
					//.connectionsPerBroker(2)

					//.statsInterval(60, TimeUnit.SECONDS)
					//.connectionTimeout(60000, TimeUnit.MILLISECONDS)
					.build();
Producer Code
userProducer = client.newProducer(Schema.BYTES)
						.topic(topic)
						.batchingMaxPublishDelay(1, TimeUnit.MILLISECONDS)
						//.batchingMaxMessages(10000) not set in rec
						.blockIfQueueFull(true)
						//.sendTimeout(10, TimeUnit.MILLISECONDS) not set in rec
						.enableBatching(true)
						.maxPendingMessages(1000)
						.maxPendingMessagesAcrossPartitions(10000)
						.batchingMaxBytes(64*1024)
						.properties(owner)
						.create();