Acknowledgement with properties for RawReader (#1006)

* Acknowledgement with properties for RawReader

This will be used by the compactor to update the topic with the
compacted ledger. There will be a cursor/subscription owned by the
compactor which keeps track of where we have compacted until, and the
ID of the ledger into which the compacted data has been written.

This change exposes the cursor's String->Long properties map which is
already available for Producer sequence tracking. This is only exposed
through the RawReader, which is part of the broker module. Clients
will not be able to use it.

* Review comments from @sijie

* Updated protobuf for recycler

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java

@@ -21,11 +21,14 @@ package org.apache.pulsar.broker.service;
 import static com.google.common.base.Preconditions.checkArgument;
 import static org.apache.pulsar.common.api.Commands.readChecksum;
 
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+import java.util.stream.Collectors;
 
 import org.apache.bookkeeper.mledger.Entry;
 import org.apache.bookkeeper.mledger.impl.PositionImpl;
@@ -251,7 +254,7 @@ public class Consumer {
                 iter.remove();
                 PositionImpl pos = (PositionImpl) entry.getPosition();
                 entry.release();
-                subscription.acknowledgeMessage(pos, AckType.Individual);
+                subscription.acknowledgeMessage(pos, AckType.Individual, Collections.emptyMap());
                 continue;
             }
             if (pendingAcks != null) {
@@ -334,15 +337,21 @@ public class Consumer {
                     position, ack.getValidationError());
         }
 
+        Map<String,Long> properties = Collections.emptyMap();
+        if (ack.getPropertiesCount() > 0) {
+            properties = ack.getPropertiesList().stream()
+                .collect(Collectors.toMap((e) -> e.getKey(),
+                                          (e) -> e.getValue()));
+        }
         if (subType == SubType.Shared) {
             // On shared subscriptions, cumulative ack is not supported
             checkArgument(ack.getAckType() == AckType.Individual);
 
             // Only ack a single message
             removePendingAcks(position);
-            subscription.acknowledgeMessage(position, AckType.Individual);
+            subscription.acknowledgeMessage(position, AckType.Individual, properties);
         } else {
-            subscription.acknowledgeMessage(position, ack.getAckType());
+            subscription.acknowledgeMessage(position, ack.getAckType(), properties);
         }
 
     }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Subscription.java

@@ -19,6 +19,7 @@
 package org.apache.pulsar.broker.service;
 
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 
 import org.apache.bookkeeper.mledger.Entry;
@@ -39,7 +40,7 @@ public interface Subscription {
 
     void consumerFlow(Consumer consumer, int additionalNumberOfMessages);
 
-    void acknowledgeMessage(PositionImpl position, AckType ackType);
+    void acknowledgeMessage(PositionImpl position, AckType ackType, Map<String,Long> properties);
 
     String getDestination();
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java

@@ -19,6 +19,7 @@
 package org.apache.pulsar.broker.service.nonpersistent;
 
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 
@@ -138,7 +139,7 @@ public class NonPersistentSubscription implements Subscription {
     }
 
     @Override
-    public void acknowledgeMessage(PositionImpl position, AckType ackType) {
+    public void acknowledgeMessage(PositionImpl position, AckType ackType, Map<String,Long> properties) {
         // No-op
     }
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

@@ -19,6 +19,7 @@
 package org.apache.pulsar.broker.service.persistent;
 
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 
@@ -175,12 +176,12 @@ public class PersistentSubscription implements Subscription {
     }
 
     @Override
-    public void acknowledgeMessage(PositionImpl position, AckType ackType) {
+    public void acknowledgeMessage(PositionImpl position, AckType ackType, Map<String,Long> properties) {
         if (ackType == AckType.Cumulative) {
             if (log.isDebugEnabled()) {
                 log.debug("[{}][{}] Cumulative ack on {}", topicName, subName, position);
             }
-            cursor.asyncMarkDelete(position, markDeleteCallback, position);
+            cursor.asyncMarkDelete(position, properties, markDeleteCallback, position);
         } else {
             if (log.isDebugEnabled()) {
                 log.debug("[{}][{}] Individual ack on {}", topicName, subName, position);

pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawReader.java

@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.client.api;
 
+import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.impl.PulsarClientImpl;
@@ -30,9 +31,9 @@ public interface RawReader {
     /**
      * Create a raw reader for a topic.
      */
-    public static CompletableFuture<RawReader> create(PulsarClient client, String topic) {
+    public static CompletableFuture<RawReader> create(PulsarClient client, String topic, String subscription) {
         CompletableFuture<Consumer> future = new CompletableFuture<>();
-        RawReader r = new RawReaderImpl((PulsarClientImpl)client, topic, future);
+        RawReader r = new RawReaderImpl((PulsarClientImpl)client, topic, subscription, future);
         return future.thenCompose((consumer) -> r.seekAsync(MessageId.earliest)).thenApply((ignore) -> r);
     }
 
@@ -49,6 +50,16 @@ public interface RawReader {
      */
     CompletableFuture<RawMessage> readNextAsync();
 
+    /**
+     * Acknowledge all messages as read up until <i>messageId</i>. The properties are stored
+     * with the individual acknowledgement, so later acknowledgements will overwrite all
+     * properties from previous acknowledgements.
+     *
+     * @param messageId to cumulatively acknowledge to
+     * @param properties a map of properties which will be stored with the acknowledgement
+     */
+    CompletableFuture<Void> acknowledgeCumulativeAsync(MessageId messageId, Map<String,Long> properties);
+
     /**
      * Close the raw reader.
      */

pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java

@@ -20,6 +20,7 @@ package org.apache.pulsar.client.impl;
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.Queue;
 import java.util.UUID;
 import java.util.concurrent.BlockingQueue;
@@ -38,6 +39,7 @@ import org.apache.pulsar.client.api.SubscriptionType;
 import org.apache.pulsar.client.impl.PulsarClientImpl;
 import org.apache.pulsar.client.impl.ConsumerImpl.SubscriptionMode;
 import org.apache.pulsar.common.api.proto.PulsarApi.MessageIdData;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
 import org.apache.pulsar.common.util.collections.GrowableArrayBlockingQueue;
 import io.netty.buffer.ByteBuf;
 import org.slf4j.Logger;
@@ -52,12 +54,12 @@ public class RawReaderImpl implements RawReader {
     private final ConsumerConfiguration consumerConfiguration;
     private RawConsumerImpl consumer;
 
-    public RawReaderImpl(PulsarClientImpl client, String topic, CompletableFuture<Consumer> consumerFuture) {
+    public RawReaderImpl(PulsarClientImpl client, String topic, String subscription,
+                         CompletableFuture<Consumer> consumerFuture) {
         this.client = client;
+        this.subscription = subscription;
         this.topic = topic;
 
-        subscription = "raw-reader";
-
         consumerConfiguration = new ConsumerConfiguration();
         consumerConfiguration.setSubscriptionType(SubscriptionType.Exclusive);
         consumerConfiguration.setReceiverQueueSize(DEFAULT_RECEIVER_QUEUE_SIZE);
@@ -76,6 +78,11 @@ public class RawReaderImpl implements RawReader {
         return consumer.receiveRawAsync();
     }
 
+    @Override
+    public CompletableFuture<Void> acknowledgeCumulativeAsync(MessageId messageId, Map<String,Long> properties) {
+        return consumer.doAcknowledge(messageId, AckType.Cumulative, properties);
+    }
+
     @Override
     public CompletableFuture<Void> closeAsync() {
         return consumer.closeAsync();

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ServerCnxTest.java

@@ -37,6 +37,7 @@ import static org.testng.Assert.assertTrue;
 import java.io.IOException;
 import java.lang.reflect.Field;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
@@ -1175,7 +1176,7 @@ public class ServerCnxTest {
         PositionImpl pos = new PositionImpl(0, 0);
 
         clientCommand = Commands.newAck(1 /* consumer id */, pos.getLedgerId(), pos.getEntryId(), AckType.Individual,
-                null);
+                                        null, Collections.emptyMap());
         channel.writeInbound(clientCommand);
 
         // verify nothing is sent out on the wire after ack

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/RawReaderTest.java

@@ -26,13 +26,19 @@ import com.google.common.collect.Sets;
 import io.netty.buffer.ByteBuf;
 
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.bookkeeper.mledger.ManagedLedger;
+
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.broker.service.persistent.PersistentTopic;
 import org.apache.pulsar.common.api.Commands;
 import org.apache.pulsar.client.api.MessageBuilder;
 import org.apache.pulsar.client.api.MessageId;
@@ -41,7 +47,6 @@ import org.apache.pulsar.client.api.ProducerConfiguration;
 import org.apache.pulsar.client.api.RawMessage;
 import org.apache.pulsar.client.api.RawReader;
 import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
-import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
 import org.apache.pulsar.common.policies.data.ClusterData;
 import org.apache.pulsar.common.policies.data.PropertyAdmin;
 
@@ -54,6 +59,7 @@ import org.testng.annotations.Test;
 
 public class RawReaderTest extends MockedPulsarServiceBaseTest {
     private static final Logger log = LoggerFactory.getLogger(RawReaderTest.class);
+    private static final String subscription = "foobar-sub";
 
     @BeforeMethod
     @Override
@@ -104,7 +110,7 @@ public class RawReaderTest extends MockedPulsarServiceBaseTest {
 
         Set<String> keys = publishMessages(topic, numKeys);
 
-        RawReader reader = RawReader.create(pulsarClient, topic).get();
+        RawReader reader = RawReader.create(pulsarClient, topic, subscription).get();
         try {
             while (true) { // should break out with TimeoutException
                 try (RawMessage m = reader.readNextAsync().get(1, TimeUnit.SECONDS)) {
@@ -126,7 +132,7 @@ public class RawReaderTest extends MockedPulsarServiceBaseTest {
         publishMessages(topic, numKeys);
 
         Set<String> readKeys = new HashSet<>();
-        RawReader reader = RawReader.create(pulsarClient, topic).get();
+        RawReader reader = RawReader.create(pulsarClient, topic, subscription).get();
         try {
             while (true) { // should break out with TimeoutException
                 try (RawMessage m = reader.readNextAsync().get(1, TimeUnit.SECONDS)) {
@@ -161,7 +167,7 @@ public class RawReaderTest extends MockedPulsarServiceBaseTest {
         publishMessages(topic, numKeys);
 
         Set<String> readKeys = new HashSet<>();
-        RawReader reader = RawReader.create(pulsarClient, topic).get();
+        RawReader reader = RawReader.create(pulsarClient, topic, subscription).get();
         int i = 0;
         MessageId seekTo = null;
         try {
@@ -206,7 +212,7 @@ public class RawReaderTest extends MockedPulsarServiceBaseTest {
 
         publishMessages(topic, numMessages);
 
-        RawReader reader = RawReader.create(pulsarClient, topic).get();
+        RawReader reader = RawReader.create(pulsarClient, topic, subscription).get();
         List<Future<RawMessage>> futures = new ArrayList<>();
         Set<String> keys = new HashSet<>();
 
@@ -226,4 +232,44 @@ public class RawReaderTest extends MockedPulsarServiceBaseTest {
         Assert.assertEquals(timeouts, 1);
         Assert.assertEquals(keys.size(), numMessages);
     }
+
+    @Test
+    public void testAcknowledgeWithProperties() throws Exception {
+        int numKeys = 10;
+
+        String topic = "persistent://my-property/use/my-ns/my-raw-topic";
+
+        Set<String> keys = publishMessages(topic, numKeys);
+
+        MessageId lastMessageId = null;
+        RawReader reader = RawReader.create(pulsarClient, topic, subscription).get();
+        try {
+            while (true) { // should break out with TimeoutException
+                try (RawMessage m = reader.readNextAsync().get(1, TimeUnit.SECONDS)) {
+                    lastMessageId = m.getMessageId();
+                    Assert.assertTrue(keys.remove(extractKey(m)));
+                }
+            }
+        } catch (TimeoutException te) {
+            // ok
+        }
+
+        Assert.assertTrue(keys.isEmpty());
+
+        Map<String,Long> properties = new HashMap<>();
+        properties.put("foobar", 0xdeadbeefdecaL);
+        reader.acknowledgeCumulativeAsync(lastMessageId, properties).get(5, TimeUnit.SECONDS);
+
+        PersistentTopic topicRef = (PersistentTopic) pulsar.getBrokerService().getTopicReference(topic);
+        ManagedLedger ledger = topicRef.getManagedLedger();
+        for (int i = 0; i < 30; i++) {
+            if (ledger.openCursor(subscription).getProperties().get("foobar") == Long.valueOf(0xdeadbeefdecaL)) {
+                break;
+            }
+            Thread.sleep(100);
+        }
+        Assert.assertEquals(ledger.openCursor(subscription).getProperties().get("foobar"),
+                Long.valueOf(0xdeadbeefdecaL));
+    }
+
 }

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java

@@ -18,6 +18,8 @@
  */
 package org.apache.pulsar.client.impl;
 
+import java.util.Collections;
+import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CompletableFuture;
@@ -225,7 +227,7 @@ public abstract class ConsumerBase extends HandlerBase implements Consumer {
 
     @Override
     public CompletableFuture<Void> acknowledgeAsync(MessageId messageId) {
-        return doAcknowledge(messageId, AckType.Individual);
+        return doAcknowledge(messageId, AckType.Individual, Collections.emptyMap());
     }
 
     @Override
@@ -235,10 +237,11 @@ public abstract class ConsumerBase extends HandlerBase implements Consumer {
                     "Cannot use cumulative acks on a non-exclusive subscription"));
         }
 
-        return doAcknowledge(messageId, AckType.Cumulative);
+        return doAcknowledge(messageId, AckType.Cumulative, Collections.emptyMap());
     }
 
-    abstract protected CompletableFuture<Void> doAcknowledge(MessageId messageId, AckType ackType);
+    abstract protected CompletableFuture<Void> doAcknowledge(MessageId messageId, AckType ackType,
+                                                             Map<String,Long> properties);
 
     @Override
     public void unsubscribe() throws PulsarClientException {

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

@@ -28,7 +28,9 @@ import static org.apache.pulsar.common.api.Commands.readChecksum;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.BitSet;
+import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.NavigableMap;
 import java.util.Objects;
 import java.util.Set;
@@ -322,7 +324,8 @@ public class ConsumerImpl extends ConsumerBase {
 
     // we may not be able to ack message being acked by client. However messages in prior
     // batch may be ackable
-    private void ackMessagesInEarlierBatch(BatchMessageIdImpl batchMessageId, MessageIdImpl message) {
+    private void ackMessagesInEarlierBatch(BatchMessageIdImpl batchMessageId, MessageIdImpl message,
+                                           Map<String,Long> properties) {
         // get entry before this message and ack that message on broker
         MessageIdImpl lowerKey = batchMessageAckTracker.lowerKey(message);
         if (lowerKey != null) {
@@ -334,7 +337,7 @@ public class ConsumerImpl extends ConsumerBase {
                 log.debug("[{}] [{}] ack prior message {} to broker on cumulative ack for message {}", subscription,
                         consumerId, lowerKey, batchMessageId);
             }
-            sendAcknowledge(lowerKey, AckType.Cumulative);
+            sendAcknowledge(lowerKey, AckType.Cumulative, properties);
         } else {
             if (log.isDebugEnabled()) {
                 log.debug("[{}] [{}] no messages prior to message {}", subscription, consumerId, batchMessageId);
@@ -342,7 +345,8 @@ public class ConsumerImpl extends ConsumerBase {
         }
     }
 
-    boolean markAckForBatchMessage(BatchMessageIdImpl batchMessageId, AckType ackType) {
+    boolean markAckForBatchMessage(BatchMessageIdImpl batchMessageId, AckType ackType,
+                                   Map<String,Long> properties) {
         // we keep track of entire batch and so need MessageIdImpl and cannot use BatchMessageIdImpl
         MessageIdImpl message = new MessageIdImpl(batchMessageId.getLedgerId(), batchMessageId.getEntryId(),
                 batchMessageId.getPartitionIndex());
@@ -396,7 +400,7 @@ public class ConsumerImpl extends ConsumerBase {
         } else {
             // we cannot ack this message to broker. but prior message may be ackable
             if (ackType == AckType.Cumulative) {
-                ackMessagesInEarlierBatch(batchMessageId, message);
+                ackMessagesInEarlierBatch(batchMessageId, message, properties);
             }
             if (log.isDebugEnabled()) {
                 log.debug("[{}] [{}] cannot ack message to broker {}, acktype {}, pending acks - {}", subscription,
@@ -439,7 +443,8 @@ public class ConsumerImpl extends ConsumerBase {
     }
 
     @Override
-    protected CompletableFuture<Void> doAcknowledge(MessageId messageId, AckType ackType) {
+    protected CompletableFuture<Void> doAcknowledge(MessageId messageId, AckType ackType,
+                                                    Map<String,Long> properties) {
         checkArgument(messageId instanceof MessageIdImpl);
         if (getState() != State.Ready && getState() != State.Connecting) {
             stats.incrementNumAcksFailed();
@@ -447,7 +452,7 @@ public class ConsumerImpl extends ConsumerBase {
         }
 
         if (messageId instanceof BatchMessageIdImpl) {
-            if (markAckForBatchMessage((BatchMessageIdImpl) messageId, ackType)) {
+            if (markAckForBatchMessage((BatchMessageIdImpl) messageId, ackType, properties)) {
                 // all messages in batch have been acked so broker can be acked via sendAcknowledge()
                 if (log.isDebugEnabled()) {
                     log.debug("[{}] [{}] acknowledging message - {}, acktype {}", subscription, consumerName, messageId,
@@ -463,12 +468,14 @@ public class ConsumerImpl extends ConsumerBase {
         if (ackType == AckType.Cumulative && !(messageId instanceof BatchMessageIdImpl)) {
             updateBatchAckTracker((MessageIdImpl) messageId, ackType);
         }
-        return sendAcknowledge(messageId, ackType);
+        return sendAcknowledge(messageId, ackType, properties);
     }
 
-    private CompletableFuture<Void> sendAcknowledge(MessageId messageId, AckType ackType) {
+    private CompletableFuture<Void> sendAcknowledge(MessageId messageId, AckType ackType,
+                                                    Map<String,Long> properties) {
         MessageIdImpl msgId = (MessageIdImpl) messageId;
-        final ByteBuf cmd = Commands.newAck(consumerId, msgId.getLedgerId(), msgId.getEntryId(), ackType, null);
+        final ByteBuf cmd = Commands.newAck(consumerId, msgId.getLedgerId(), msgId.getEntryId(),
+                                            ackType, null, properties);
 
         // There's no actual response from ack messages
         final CompletableFuture<Void> ackFuture = new CompletableFuture<Void>();
@@ -1094,7 +1101,7 @@ public class ConsumerImpl extends ConsumerBase {
 
     private void discardMessage(MessageIdData messageId, ClientCnx currentCnx, ValidationError validationError) {
         ByteBuf cmd = Commands.newAck(consumerId, messageId.getLedgerId(), messageId.getEntryId(), AckType.Individual,
-                validationError);
+                                      validationError, Collections.emptyMap());
         currentCnx.ctx().writeAndFlush(cmd, currentCnx.ctx().voidPromise());
         increaseAvailablePermits(currentCnx);
         stats.incrementNumReceiveFailed();

pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedConsumerImpl.java

@@ -236,7 +236,8 @@ public class PartitionedConsumerImpl extends ConsumerBase {
     }
 
     @Override
-    protected CompletableFuture<Void> doAcknowledge(MessageId messageId, AckType ackType) {
+    protected CompletableFuture<Void> doAcknowledge(MessageId messageId, AckType ackType,
+                                                    Map<String,Long> properties) {
         checkArgument(messageId instanceof MessageIdImpl);
 
         if (getState() != State.Ready) {
@@ -249,7 +250,7 @@ public class PartitionedConsumerImpl extends ConsumerBase {
         } else {
 
             ConsumerImpl consumer = consumers.get(((MessageIdImpl) messageId).getPartitionIndex());
-            return consumer.doAcknowledge(messageId, ackType).thenRun(() ->
+            return consumer.doAcknowledge(messageId, ackType, properties).thenRun(() ->
                     unAckedMessageTracker.remove((MessageIdImpl) messageId));
         }
 

pulsar-common/src/main/java/org/apache/pulsar/common/api/Commands.java

@@ -23,6 +23,7 @@ import static org.apache.pulsar.checksum.utils.Crc32cChecksum.resumeChecksum;
 
 import java.io.IOException;
 import java.util.List;
+import java.util.Map;
 
 import org.apache.pulsar.common.api.proto.PulsarApi;
 import org.apache.pulsar.common.api.proto.PulsarApi.AuthMethod;
@@ -517,7 +518,7 @@ public class Commands {
     }
 
     public static ByteBuf newAck(long consumerId, long ledgerId, long entryId, AckType ackType,
-            ValidationError validationError) {
+                                 ValidationError validationError, Map<String,Long> properties) {
         CommandAck.Builder ackBuilder = CommandAck.newBuilder();
         ackBuilder.setConsumerId(consumerId);
         ackBuilder.setAckType(ackType);
@@ -529,6 +530,10 @@ public class Commands {
         if (validationError != null) {
             ackBuilder.setValidationError(validationError);
         }
+        for (Map.Entry<String,Long> e : properties.entrySet()) {
+            ackBuilder.addProperties(
+                    PulsarApi.KeyLongValue.newBuilder().setKey(e.getKey()).setValue(e.getValue()).build());
+        }
         CommandAck ack = ackBuilder.build();
 
         ByteBuf res = serializeWithSize(BaseCommand.newBuilder().setType(Type.ACK).setAck(ack));

pulsar-common/src/main/java/org/apache/pulsar/common/api/proto/PulsarApi.java

@@ -1225,6 +1225,438 @@ public final class PulsarApi {
     // @@protoc_insertion_point(class_scope:pulsar.proto.KeyValue)
   }
   
+  public interface KeyLongValueOrBuilder
+      extends com.google.protobuf.MessageLiteOrBuilder {
+    
+    // required string key = 1;
+    boolean hasKey();
+    String getKey();
+    
+    // required uint64 value = 2;
+    boolean hasValue();
+    long getValue();
+  }
+  public static final class KeyLongValue extends
+      com.google.protobuf.GeneratedMessageLite
+      implements KeyLongValueOrBuilder, org.apache.pulsar.common.util.protobuf.ByteBufCodedOutputStream.ByteBufGeneratedMessage  {
+    // Use KeyLongValue.newBuilder() to construct.
+    private final io.netty.util.Recycler.Handle<KeyLongValue> handle;
+    private KeyLongValue(io.netty.util.Recycler.Handle<KeyLongValue> handle) {
+      this.handle = handle;
+    }
+    
+     private static final io.netty.util.Recycler<KeyLongValue> RECYCLER = new io.netty.util.Recycler<KeyLongValue>() {
+            protected KeyLongValue newObject(Handle<KeyLongValue> handle) {
+              return new KeyLongValue(handle);
+            }
+          };
+        
+        public void recycle() {
+            this.initFields();
+            this.memoizedIsInitialized = -1;
+            this.bitField0_ = 0;
+            this.memoizedSerializedSize = -1;
+            handle.recycle(this);
+        }
+         
+    private KeyLongValue(boolean noInit) {
+        this.handle = null;
+    }
+    
+    private static final KeyLongValue defaultInstance;
+    public static KeyLongValue getDefaultInstance() {
+      return defaultInstance;
+    }
+    
+    public KeyLongValue getDefaultInstanceForType() {
+      return defaultInstance;
+    }
+    
+    private int bitField0_;
+    // required string key = 1;
+    public static final int KEY_FIELD_NUMBER = 1;
+    private java.lang.Object key_;
+    public boolean hasKey() {
+      return ((bitField0_ & 0x00000001) == 0x00000001);
+    }
+    public String getKey() {
+      java.lang.Object ref = key_;
+      if (ref instanceof String) {
+        return (String) ref;
+      } else {
+        com.google.protobuf.ByteString bs = 
+            (com.google.protobuf.ByteString) ref;
+        String s = bs.toStringUtf8();
+        if (com.google.protobuf.Internal.isValidUtf8(bs)) {
+          key_ = s;
+        }
+        return s;
+      }
+    }
+    private com.google.protobuf.ByteString getKeyBytes() {
+      java.lang.Object ref = key_;
+      if (ref instanceof String) {
+        com.google.protobuf.ByteString b = 
+            com.google.protobuf.ByteString.copyFromUtf8((String) ref);
+        key_ = b;
+        return b;
+      } else {
+        return (com.google.protobuf.ByteString) ref;
+      }
+    }
+    
+    // required uint64 value = 2;
+    public static final int VALUE_FIELD_NUMBER = 2;
+    private long value_;
+    public boolean hasValue() {
+      return ((bitField0_ & 0x00000002) == 0x00000002);
+    }
+    public long getValue() {
+      return value_;
+    }
+    
+    private void initFields() {
+      key_ = "";
+      value_ = 0L;
+    }
+    private byte memoizedIsInitialized = -1;
+    public final boolean isInitialized() {
+      byte isInitialized = memoizedIsInitialized;
+      if (isInitialized != -1) return isInitialized == 1;
+      
+      if (!hasKey()) {
+        memoizedIsInitialized = 0;
+        return false;
+      }
+      if (!hasValue()) {
+        memoizedIsInitialized = 0;
+        return false;
+      }
+      memoizedIsInitialized = 1;
+      return true;
+    }
+    
+    public void writeTo(com.google.protobuf.CodedOutputStream output)
+                        throws java.io.IOException {
+        throw new RuntimeException("Cannot use CodedOutputStream");
+    }
+    
+    public void writeTo(org.apache.pulsar.common.util.protobuf.ByteBufCodedOutputStream output)
+                        throws java.io.IOException {
+      getSerializedSize();
+      if (((bitField0_ & 0x00000001) == 0x00000001)) {
+        output.writeBytes(1, getKeyBytes());
+      }
+      if (((bitField0_ & 0x00000002) == 0x00000002)) {
+        output.writeUInt64(2, value_);
+      }
+    }
+    
+    private int memoizedSerializedSize = -1;
+    public int getSerializedSize() {
+      int size = memoizedSerializedSize;
+      if (size != -1) return size;
+    
+      size = 0;
+      if (((bitField0_ & 0x00000001) == 0x00000001)) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeBytesSize(1, getKeyBytes());
+      }
+      if (((bitField0_ & 0x00000002) == 0x00000002)) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeUInt64Size(2, value_);
+      }
+      memoizedSerializedSize = size;
+      return size;
+    }
+    
+    private static final long serialVersionUID = 0L;
+    @java.lang.Override
+    protected java.lang.Object writeReplace()
+        throws java.io.ObjectStreamException {
+      return super.writeReplace();
+    }
+    
+    public static org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue parseFrom(
+        com.google.protobuf.ByteString data)
+        throws com.google.protobuf.InvalidProtocolBufferException {
+         throw new RuntimeException("Disabled");
+    }
+    public static org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue parseFrom(
+        com.google.protobuf.ByteString data,
+        com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+        throws com.google.protobuf.InvalidProtocolBufferException {
+         throw new RuntimeException("Disabled");
+    }
+    public static org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue parseFrom(byte[] data)
+        throws com.google.protobuf.InvalidProtocolBufferException {
+      return newBuilder().mergeFrom(data).buildParsed();
+    }
+    public static org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue parseFrom(
+        byte[] data,
+        com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+        throws com.google.protobuf.InvalidProtocolBufferException {
+      return newBuilder().mergeFrom(data, extensionRegistry)
+               .buildParsed();
+    }
+    public static org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue parseFrom(java.io.InputStream input)
+        throws java.io.IOException {
+      return newBuilder().mergeFrom(input).buildParsed();
+    }
+    public static org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue parseFrom(
+        java.io.InputStream input,
+        com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+        throws java.io.IOException {
+      return newBuilder().mergeFrom(input, extensionRegistry)
+               .buildParsed();
+    }
+    public static org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue parseDelimitedFrom(java.io.InputStream input)
+        throws java.io.IOException {
+      Builder builder = newBuilder();
+      if (builder.mergeDelimitedFrom(input)) {
+        return builder.buildParsed();
+      } else {
+        return null;
+      }
+    }
+    public static org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue parseDelimitedFrom(
+        java.io.InputStream input,
+        com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+        throws java.io.IOException {
+      Builder builder = newBuilder();
+      if (builder.mergeDelimitedFrom(input, extensionRegistry)) {
+        return builder.buildParsed();
+      } else {
+        return null;
+      }
+    }
+    public static org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue parseFrom(
+        com.google.protobuf.CodedInputStream input)
+        throws java.io.IOException {
+      return newBuilder().mergeFrom(input).buildParsed();
+    }
+    public static org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue parseFrom(
+        com.google.protobuf.CodedInputStream input,
+        com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+        throws java.io.IOException {
+      return newBuilder().mergeFrom(input, extensionRegistry)
+               .buildParsed();
+    }
+    
+    public static Builder newBuilder() { return Builder.create(); }
+    public Builder newBuilderForType() { return newBuilder(); }
+    public static Builder newBuilder(org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue prototype) {
+      return newBuilder().mergeFrom(prototype);
+    }
+    public Builder toBuilder() { return newBuilder(this); }
+    
+    public static final class Builder extends
+        com.google.protobuf.GeneratedMessageLite.Builder<
+          org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue, Builder>
+        implements org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValueOrBuilder, org.apache.pulsar.common.util.protobuf.ByteBufCodedInputStream.ByteBufMessageBuilder  {
+      // Construct using org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue.newBuilder()
+      private final io.netty.util.Recycler.Handle<Builder> handle;
+      private Builder(io.netty.util.Recycler.Handle<Builder> handle) {
+        this.handle = handle;
+        maybeForceBuilderInitialization();
+      }
+      private final static io.netty.util.Recycler<Builder> RECYCLER = new io.netty.util.Recycler<Builder>() {
+         protected Builder newObject(io.netty.util.Recycler.Handle<Builder> handle) {
+               return new Builder(handle);
+             }
+            };
+      
+       public void recycle() {
+                clear();
+                handle.recycle(this);
+            }
+      
+      private void maybeForceBuilderInitialization() {
+      }
+      private static Builder create() {
+        return RECYCLER.get();
+      }
+      
+      public Builder clear() {
+        super.clear();
+        key_ = "";
+        bitField0_ = (bitField0_ & ~0x00000001);
+        value_ = 0L;
+        bitField0_ = (bitField0_ & ~0x00000002);
+        return this;
+      }
+      
+      public Builder clone() {
+        return create().mergeFrom(buildPartial());
+      }
+      
+      public org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue getDefaultInstanceForType() {
+        return org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue.getDefaultInstance();
+      }
+      
+      public org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue build() {
+        org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue result = buildPartial();
+        if (!result.isInitialized()) {
+          throw newUninitializedMessageException(result);
+        }
+        return result;
+      }
+      
+      private org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue buildParsed()
+          throws com.google.protobuf.InvalidProtocolBufferException {
+        org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue result = buildPartial();
+        if (!result.isInitialized()) {
+          throw newUninitializedMessageException(
+            result).asInvalidProtocolBufferException();
+        }
+        return result;
+      }
+      
+      public org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue buildPartial() {
+        org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue result = org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue.RECYCLER.get();
+        int from_bitField0_ = bitField0_;
+        int to_bitField0_ = 0;
+        if (((from_bitField0_ & 0x00000001) == 0x00000001)) {
+          to_bitField0_ |= 0x00000001;
+        }
+        result.key_ = key_;
+        if (((from_bitField0_ & 0x00000002) == 0x00000002)) {
+          to_bitField0_ |= 0x00000002;
+        }
+        result.value_ = value_;
+        result.bitField0_ = to_bitField0_;
+        return result;
+      }
+      
+      public Builder mergeFrom(org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue other) {
+        if (other == org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue.getDefaultInstance()) return this;
+        if (other.hasKey()) {
+          setKey(other.getKey());
+        }
+        if (other.hasValue()) {
+          setValue(other.getValue());
+        }
+        return this;
+      }
+      
+      public final boolean isInitialized() {
+        if (!hasKey()) {
+          
+          return false;
+        }
+        if (!hasValue()) {
+          
+          return false;
+        }
+        return true;
+      }
+      
+      public Builder mergeFrom(com.google.protobuf.CodedInputStream input,
+                              com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+                              throws java.io.IOException {
+         throw new java.io.IOException("Merge from CodedInputStream is disabled");
+                              }
+      public Builder mergeFrom(
+          org.apache.pulsar.common.util.protobuf.ByteBufCodedInputStream input,
+          com.google.protobuf.ExtensionRegistryLite extensionRegistry)
+          throws java.io.IOException {
+        while (true) {
+          int tag = input.readTag();
+          switch (tag) {
+            case 0:
+              
+              return this;
+            default: {
+              if (!input.skipField(tag)) {
+                
+                return this;
+              }
+              break;
+            }
+            case 10: {
+              bitField0_ |= 0x00000001;
+              key_ = input.readBytes();
+              break;
+            }
+            case 16: {
+              bitField0_ |= 0x00000002;
+              value_ = input.readUInt64();
+              break;
+            }
+          }
+        }
+      }
+      
+      private int bitField0_;
+      
+      // required string key = 1;
+      private java.lang.Object key_ = "";
+      public boolean hasKey() {
+        return ((bitField0_ & 0x00000001) == 0x00000001);
+      }
+      public String getKey() {
+        java.lang.Object ref = key_;
+        if (!(ref instanceof String)) {
+          String s = ((com.google.protobuf.ByteString) ref).toStringUtf8();
+          key_ = s;
+          return s;
+        } else {
+          return (String) ref;
+        }
+      }
+      public Builder setKey(String value) {
+        if (value == null) {
+    throw new NullPointerException();
+  }
+  bitField0_ |= 0x00000001;
+        key_ = value;
+        
+        return this;
+      }
+      public Builder clearKey() {
+        bitField0_ = (bitField0_ & ~0x00000001);
+        key_ = getDefaultInstance().getKey();
+        
+        return this;
+      }
+      void setKey(com.google.protobuf.ByteString value) {
+        bitField0_ |= 0x00000001;
+        key_ = value;
+        
+      }
+      
+      // required uint64 value = 2;
+      private long value_ ;
+      public boolean hasValue() {
+        return ((bitField0_ & 0x00000002) == 0x00000002);
+      }
+      public long getValue() {
+        return value_;
+      }
+      public Builder setValue(long value) {
+        bitField0_ |= 0x00000002;
+        value_ = value;
+        
+        return this;
+      }
+      public Builder clearValue() {
+        bitField0_ = (bitField0_ & ~0x00000002);
+        value_ = 0L;
+        
+        return this;
+      }
+      
+      // @@protoc_insertion_point(builder_scope:pulsar.proto.KeyLongValue)
+    }
+    
+    static {
+      defaultInstance = new KeyLongValue(true);
+      defaultInstance.initFields();
+    }
+    
+    // @@protoc_insertion_point(class_scope:pulsar.proto.KeyLongValue)
+  }
+  
   public interface EncryptionKeysOrBuilder
       extends com.google.protobuf.MessageLiteOrBuilder {
     
@@ -11255,6 +11687,12 @@ public final class PulsarApi {
     // optional .pulsar.proto.CommandAck.ValidationError validation_error = 4;
     boolean hasValidationError();
     org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.ValidationError getValidationError();
+    
+    // repeated .pulsar.proto.KeyLongValue properties = 5;
+    java.util.List<org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue> 
+        getPropertiesList();
+    org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue getProperties(int index);
+    int getPropertiesCount();
   }
   public static final class CommandAck extends
       com.google.protobuf.GeneratedMessageLite
@@ -11424,11 +11862,33 @@ public final class PulsarApi {
       return validationError_;
     }
     
+    // repeated .pulsar.proto.KeyLongValue properties = 5;
+    public static final int PROPERTIES_FIELD_NUMBER = 5;
+    private java.util.List<org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue> properties_;
+    public java.util.List<org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue> getPropertiesList() {
+      return properties_;
+    }
+    public java.util.List<? extends org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValueOrBuilder> 
+        getPropertiesOrBuilderList() {
+      return properties_;
+    }
+    public int getPropertiesCount() {
+      return properties_.size();
+    }
+    public org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue getProperties(int index) {
+      return properties_.get(index);
+    }
+    public org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValueOrBuilder getPropertiesOrBuilder(
+        int index) {
+      return properties_.get(index);
+    }
+    
     private void initFields() {
       consumerId_ = 0L;
       ackType_ = org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType.Individual;
       messageId_ = org.apache.pulsar.common.api.proto.PulsarApi.MessageIdData.getDefaultInstance();
       validationError_ = org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.ValidationError.UncompressedSizeCorruption;
+      properties_ = java.util.Collections.emptyList();
     }
     private byte memoizedIsInitialized = -1;
     public final boolean isInitialized() {
@@ -11451,6 +11911,12 @@ public final class PulsarApi {
         memoizedIsInitialized = 0;
         return false;
       }
+      for (int i = 0; i < getPropertiesCount(); i++) {
+        if (!getProperties(i).isInitialized()) {
+          memoizedIsInitialized = 0;
+          return false;
+        }
+      }
       memoizedIsInitialized = 1;
       return true;
     }
@@ -11475,6 +11941,9 @@ public final class PulsarApi {
       if (((bitField0_ & 0x00000008) == 0x00000008)) {
         output.writeEnum(4, validationError_.getNumber());
       }
+      for (int i = 0; i < properties_.size(); i++) {
+        output.writeMessage(5, properties_.get(i));
+      }
     }
     
     private int memoizedSerializedSize = -1;
@@ -11499,6 +11968,10 @@ public final class PulsarApi {
         size += com.google.protobuf.CodedOutputStream
           .computeEnumSize(4, validationError_.getNumber());
       }
+      for (int i = 0; i < properties_.size(); i++) {
+        size += com.google.protobuf.CodedOutputStream
+          .computeMessageSize(5, properties_.get(i));
+      }
       memoizedSerializedSize = size;
       return size;
     }
@@ -11620,6 +12093,8 @@ public final class PulsarApi {
         bitField0_ = (bitField0_ & ~0x00000004);
         validationError_ = org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.ValidationError.UncompressedSizeCorruption;
         bitField0_ = (bitField0_ & ~0x00000008);
+        properties_ = java.util.Collections.emptyList();
+        bitField0_ = (bitField0_ & ~0x00000010);
         return this;
       }
       
@@ -11669,6 +12144,11 @@ public final class PulsarApi {
           to_bitField0_ |= 0x00000008;
         }
         result.validationError_ = validationError_;
+        if (((bitField0_ & 0x00000010) == 0x00000010)) {
+          properties_ = java.util.Collections.unmodifiableList(properties_);
+          bitField0_ = (bitField0_ & ~0x00000010);
+        }
+        result.properties_ = properties_;
         result.bitField0_ = to_bitField0_;
         return result;
       }
@@ -11687,6 +12167,16 @@ public final class PulsarApi {
         if (other.hasValidationError()) {
           setValidationError(other.getValidationError());
         }
+        if (!other.properties_.isEmpty()) {
+          if (properties_.isEmpty()) {
+            properties_ = other.properties_;
+            bitField0_ = (bitField0_ & ~0x00000010);
+          } else {
+            ensurePropertiesIsMutable();
+            properties_.addAll(other.properties_);
+          }
+          
+        }
         return this;
       }
       
@@ -11707,6 +12197,12 @@ public final class PulsarApi {
           
           return false;
         }
+        for (int i = 0; i < getPropertiesCount(); i++) {
+          if (!getProperties(i).isInitialized()) {
+            
+            return false;
+          }
+        }
         return true;
       }
       
@@ -11765,6 +12261,12 @@ public final class PulsarApi {
               }
               break;
             }
+            case 42: {
+              org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue.Builder subBuilder = org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue.newBuilder();
+              input.readMessage(subBuilder, extensionRegistry);
+              addProperties(subBuilder.buildPartial());
+              break;
+            }
           }
         }
       }
@@ -11883,6 +12385,95 @@ public final class PulsarApi {
         return this;
       }
       
+      // repeated .pulsar.proto.KeyLongValue properties = 5;
+      private java.util.List<org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue> properties_ =
+        java.util.Collections.emptyList();
+      private void ensurePropertiesIsMutable() {
+        if (!((bitField0_ & 0x00000010) == 0x00000010)) {
+          properties_ = new java.util.ArrayList<org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue>(properties_);
+          bitField0_ |= 0x00000010;
+         }
+      }
+      
+      public java.util.List<org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue> getPropertiesList() {
+        return java.util.Collections.unmodifiableList(properties_);
+      }
+      public int getPropertiesCount() {
+        return properties_.size();
+      }
+      public org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue getProperties(int index) {
+        return properties_.get(index);
+      }
+      public Builder setProperties(
+          int index, org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue value) {
+        if (value == null) {
+          throw new NullPointerException();
+        }
+        ensurePropertiesIsMutable();
+        properties_.set(index, value);
+        
+        return this;
+      }
+      public Builder setProperties(
+          int index, org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue.Builder builderForValue) {
+        ensurePropertiesIsMutable();
+        properties_.set(index, builderForValue.build());
+        
+        return this;
+      }
+      public Builder addProperties(org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue value) {
+        if (value == null) {
+          throw new NullPointerException();
+        }
+        ensurePropertiesIsMutable();
+        properties_.add(value);
+        
+        return this;
+      }
+      public Builder addProperties(
+          int index, org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue value) {
+        if (value == null) {
+          throw new NullPointerException();
+        }
+        ensurePropertiesIsMutable();
+        properties_.add(index, value);
+        
+        return this;
+      }
+      public Builder addProperties(
+          org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue.Builder builderForValue) {
+        ensurePropertiesIsMutable();
+        properties_.add(builderForValue.build());
+        
+        return this;
+      }
+      public Builder addProperties(
+          int index, org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue.Builder builderForValue) {
+        ensurePropertiesIsMutable();
+        properties_.add(index, builderForValue.build());
+        
+        return this;
+      }
+      public Builder addAllProperties(
+          java.lang.Iterable<? extends org.apache.pulsar.common.api.proto.PulsarApi.KeyLongValue> values) {
+        ensurePropertiesIsMutable();
+        super.addAll(values, properties_);
+        
+        return this;
+      }
+      public Builder clearProperties() {
+        properties_ = java.util.Collections.emptyList();
+        bitField0_ = (bitField0_ & ~0x00000010);
+        
+        return this;
+      }
+      public Builder removeProperties(int index) {
+        ensurePropertiesIsMutable();
+        properties_.remove(index);
+        
+        return this;
+      }
+      
       // @@protoc_insertion_point(builder_scope:pulsar.proto.CommandAck)
     }
     

pulsar-common/src/main/proto/PulsarApi.proto

@@ -34,6 +34,11 @@ message KeyValue {
 	required string value = 2;
 }
 
+message KeyLongValue {
+        required string key = 1;
+        required uint64 value = 2;
+}
+
 message EncryptionKeys {
 	required string key = 1;
 	required bytes value = 2;
@@ -280,6 +285,7 @@ message CommandAck {
 	}
 
 	optional ValidationError validation_error = 4;
+        repeated KeyLongValue properties = 5;
 }
 
 message CommandFlow {