resolved comments

Author: Heesung Sohn

lib/ClientConnection.cc

@@ -1794,6 +1794,7 @@ void ClientConnection::handleTopicMigrated(const proto::CommandTopicMigrated& co
         return;
     }
 
+    Lock lock(mutex_);
     if (commandTopicMigrated.resource_type() == proto::CommandTopicMigrated_ResourceType_Producer) {
         auto it = producers_.find(resourceId);
         if (it != producers_.end()) {

lib/ProducerImpl.cc

@@ -144,8 +144,8 @@ Future<Result, bool> ProducerImpl::connectionOpened(const ClientConnectionPtr& c
         return promise.getFuture();
     }
 
-    LOG_DEBUG("Creating producer" << topic() << "producerName: " << producerName_ << " on "
-                                  << cnx->cnxString());
+    LOG_DEBUG("Creating producer for topic:" << topic() << ", producerName:" << producerName_ << " on "
+                                             << cnx->cnxString());
     ClientImplPtr client = client_.lock();
     cnx->registerProducer(producerId_, shared_from_this());
 

proto/PulsarApi.proto

@@ -622,7 +622,7 @@ message CommandReachedEndOfTopic {
 }
 
 message CommandTopicMigrated {
-	enum ResourceType {
+    enum ResourceType {
         Producer = 0;
         Consumer = 1;
     }

tests/extensibleLM/ExtensibleLoadManagerTest.cc

@@ -70,7 +70,9 @@ TEST(ExtensibleLoadManagerTest, testPubSubWhileUnloading) {
         return res == 204 || res == 409;
     }));
 
-    Client client{"pulsar://localhost:6650"};
+    ClientConfiguration conf;
+    conf.setIOThreads(8);
+    Client client{"pulsar://localhost:6650", conf};
     Producer producer;
     ProducerConfiguration producerConfiguration;
     Result producerResult = client.createProducer(topicName, producerConfiguration, producer);