[fix] Fix PartitionedProducerImpl::closeAsync to close sub-producers properly

[erobot]

Motivation

PartitionedProducerImpl do not close sub-producers properly when any sub-producer creation fails. Continuing to retry creating producer will eventually reach the maximum producer limit. It seems a regression caused by #54.

When sub-producer creation fails, state_ is set to Failed. PartitionedProducerImpl::closeAsync only do cleanup when state_==Ready and sub-producers do not close when state_==Failed.

pulsar-client-cpp/lib/PartitionedProducerImpl.cc

Lines 273 to 276 in f0268ec

	State expectedState = Ready;
	if (!state_.compare_exchange_strong(expectedState, Closing)) {
	return;
	}

Modifications

Close sub-producers when state != Closed.

Verifying this change

test code:

Client client("pulsar://localhost:6650");
Producer producer;
while (true) {
    Result result = client.createProducer("persistent://public/default/test", producer);
    if (result != ResultOk) {
        cerr << "Error creating producer: " << result << endl;
        std::this_thread::sleep_for(std::chrono::seconds(1));
        continue;
    }
    break;
}

test cmd:

pulsar-admin namespaces set-max-producers-per-topic -p 10 public/default
pulsar-admin topics create-partitioned-topic -n 2 public/default/test
pulsar-perf produce -r 1 -n 10 public/default/test-partition-0

test procedure:

1. set namespace max producer limit to 10
2. create partitoned topic public/default/test with 2 partitions
3. use pulsar-perf to create 10 prodcuers on test-partition-0
4. run test code to create producer for public/default/test, will fail because test-partition-0 already have 10 producer
5. • without fix, test-partition-1 will have 10 producers because of sub-producer not close
   • with fix, test-partition-1 will have not more than 1 producer

Documentation

• doc-required
  (Your PR needs to update docs and you will update later)

• doc-not-needed
  bug-fix only

• doc
  (Your PR contains doc changes)

• doc-complete
  (Docs have been already added)

[shibd]

@erobot Hi, Can you add unit test based on the stepsverifying this change chapter?

[BewareMyPower]

Could you also add the unit test to avoid the regression?

[erobot]

ok, I will fix the code style and add a unit test latter.

[BewareMyPower]

without fix, test-partition-1 will have 10 producers because of sub-producer not close
with fix, test-partition-1 will have not more than 1 producer

The difference is not clear from the output.

with-fix.log

without-fix.log

Maybe a unit test would be better to tell the difference.

[BewareMyPower]

Just digged deeper into this issue. I think there is no problem with the current master.

If one internal producer failed to create, the partitioned producer would not be added into the ClientImpl or ClientConnection, see https://github.com/apache/pulsar-client-cpp/blob/96d870527d2a078645c4866e9d75b3f3b97193ba/lib/ClientImpl.cc#L202-204

Then the PartitionedProducerImpl will be destroyed and shutdown() will be called for the cleanup work (cancel the timer) in the destructor. Regarding the ProducerImpl objects, since they also failed to create, there is no need to call closeAsync on them.

1. These producers are never reachable. The destructor could handle the cleanup work well.
2. Since the creations of them are also failed, the broker side didn't register these producers, closing them makes no difference.

[erobot]

@BewareMyPower add a test to reproduce the problem

[BewareMyPower]

I got it. It's caused when some sub-producers can be created successfully while other could fail.

[BewareMyPower]

typo