eolivelli commented on a change in pull request #10279:
URL: https://github.com/apache/pulsar/pull/10279#discussion_r619978186
##########
File path:
pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
##########
@@ -1521,6 +1522,17 @@ public void testMaxTopicsPerNamespace() throws Exception
{
}
// check producer/consumer auto create partitioned topic
+ final Function<Producer<byte[]>, Producer<byte[]>> send = (p) -> {
Review comment:
Why are you modifying an existing test?
It is okay to refactor and reuse an existing test but in this case probably
you are altering the behaviour of the existing guest and also you are not sure
that you are testing your feature properly
##########
File path:
pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java
##########
@@ -169,6 +186,30 @@ private void start() {
@Override
CompletableFuture<MessageId> internalSendWithTxnAsync(Message<?> message,
Transaction txn) {
+ int partition = routerPolicy.choosePartition(message, topicMetadata);
+ checkArgument(partition >= 0 && partition <
topicMetadata.numPartitions(),
+ "Illegal partition index chosen by the message routing policy:
" + partition);
+
+ if (!producers.containsKey(partition)) {
+ final State createState = createProducer(partition).handle((prod,
createException) -> {
+ if (createException != null) {
+ log.error("[{}] Could not create internal producer.
partitionIndex: {}", topic, partition,
+ createException);
+ try {
+ producers.remove(partition).close();
Review comment:
We are not sure that we are removing the newly create producer here.
--
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.
For queries about this service, please contact Infrastructure at:
[email protected]
