[GitHub] [pulsar] merlimat commented on a change in pull request #7198: [pulsar-client-cpp] Support Seek on Partitioned Topic

[GitBox <gi...@apache.org>]

merlimat commented on a change in pull request #7198:
URL: https://github.com/apache/pulsar/pull/7198#discussion_r436455202



##########
File path: pulsar-client-cpp/lib/PartitionedConsumerImpl.cc
##########
@@ -542,11 +542,27 @@ void PartitionedConsumerImpl::handleGetConsumerStats(Result res, BrokerConsumerS
 }
 
 void PartitionedConsumerImpl::seekAsync(const MessageId& msgId, ResultCallback callback) {
-    callback(ResultOperationNotSupported);
+    Lock stateLock(mutex_);
+    if (state_ != Ready) {
+        callback(ResultAlreadyClosed);
+        return;
+    }
+    stateLock.unlock();
+    for (ConsumerList::const_iterator i = consumers_.begin(); i != consumers_.end(); i++) {
+        (*i)->seekAsync(msgId, callback);

Review comment:
       The message id is only relative to 1 partition. Seeking with same message id on multiple partition will position the subscription in the wrong position.

##########
File path: pulsar-client-cpp/lib/PartitionedConsumerImpl.cc
##########
@@ -542,11 +542,27 @@ void PartitionedConsumerImpl::handleGetConsumerStats(Result res, BrokerConsumerS
 }
 
 void PartitionedConsumerImpl::seekAsync(const MessageId& msgId, ResultCallback callback) {
-    callback(ResultOperationNotSupported);
+    Lock stateLock(mutex_);
+    if (state_ != Ready) {
+        callback(ResultAlreadyClosed);
+        return;
+    }
+    stateLock.unlock();
+    for (ConsumerList::const_iterator i = consumers_.begin(); i != consumers_.end(); i++) {
+        (*i)->seekAsync(msgId, callback);
+    }
 }
 
 void PartitionedConsumerImpl::seekAsync(uint64_t timestamp, ResultCallback callback) {
-    callback(ResultOperationNotSupported);
+    Lock stateLock(mutex_);
+    if (state_ != Ready) {
+        callback(ResultAlreadyClosed);

Review comment:
       The mutex should be released before triggering the callback, since that might be blocking




----------------------------------------------------------------
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:
users@infra.apache.org