[GitHub] zhaijack commented on a change in pull request #1066: Issue 937: add CommandGetLastMessageId to make reader know the end of topic

[GitBox]

zhaijack commented on a change in pull request #1066: Issue 937: add 
CommandGetLastMessageId to make reader know the end of topic
URL: https://github.com/apache/incubator-pulsar/pull/1066#discussion_r168061308
 
 

 ##
 File path: 
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
 ##
 @@ -1248,6 +1256,102 @@ public void seek(MessageId messageId) throws 
PulsarClientException {
 return seekFuture;
 }
 
+public boolean hasMessageAvailable() throws PulsarClientException {
+try {
+if (lastMessageIdInBroker.compareTo(lastDequeuedMessage) > 0 &&
+((MessageIdImpl)lastMessageIdInBroker).getEntryId() != -1) {
+return true;
+}
+
+return hasMessageAvailableAsync().get();
+} catch (ExecutionException | InterruptedException e) {
+throw new PulsarClientException(e);
+}
+}
+
+public CompletableFuture hasMessageAvailableAsync() {
+final CompletableFuture booleanFuture = new 
CompletableFuture<>();
+
+if (lastMessageIdInBroker.compareTo(lastDequeuedMessage) > 0 &&
+((MessageIdImpl)lastMessageIdInBroker).getEntryId() != -1) {
+booleanFuture.complete(true);
+} else {
+getLastMessageIdAsync().thenAccept(messageId -> {
+lastMessageIdInBroker = messageId;
+if (lastMessageIdInBroker.compareTo(lastDequeuedMessage) > 0 &&
+((MessageIdImpl)lastMessageIdInBroker).getEntryId() != -1) 
{
+booleanFuture.complete(true);
+} else {
+booleanFuture.complete(false);
+}
+}).exceptionally(e -> {
+log.error("[{}][{}] Failed getLastMessageId command", topic, 
subscription);
+booleanFuture.completeExceptionally(e.getCause());
+return null;
+});
+}
+return booleanFuture;
+}
+
+private CompletableFuture getLastMessageIdAsync() {
+if (getState() == State.Closing || getState() == State.Closed) {
+return FutureUtil
+.failedFuture(new 
PulsarClientException.AlreadyClosedException("Consumer was already closed"));
+}
+
+AtomicLong opTimeoutMs = new 
AtomicLong(client.getConfiguration().getOperationTimeoutMs());
+Backoff backoff = new Backoff(100, TimeUnit.MILLISECONDS,
+opTimeoutMs.get() * 2, TimeUnit.MILLISECONDS,
+0 , TimeUnit.MILLISECONDS);
+CompletableFuture getLastMessageIdFuture = new 
CompletableFuture<>();
+
+internalGetLastMessageIdAsync(backoff, opTimeoutMs, 
getLastMessageIdFuture);
+return getLastMessageIdFuture;
+}
+
+private void internalGetLastMessageIdAsync(final Backoff backoff,
+   final AtomicLong remainingTime,
+   CompletableFuture 
future) {
+if (isConnected()) {
 
 Review comment:
   Thanks, will change it.


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] zhaijack commented on a change in pull request #1066: Issue 937: add CommandGetLastMessageId to make reader know the end of topic

[GitBox]

zhaijack commented on a change in pull request #1066: Issue 937: add 
CommandGetLastMessageId to make reader know the end of topic
URL: https://github.com/apache/incubator-pulsar/pull/1066#discussion_r168036825
 
 

 ##
 File path: 
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
 ##
 @@ -1248,6 +1254,103 @@ public void seek(MessageId messageId) throws 
PulsarClientException {
 return seekFuture;
 }
 
+public boolean hasMessageAvailable() throws PulsarClientException {
+try {
+if (lastMessageIdInBroker.compareTo(lastDequeuedMessage) > 0 &&
+((MessageIdImpl)lastMessageIdInBroker).getEntryId() != -1) {
+return true;
+}
+
+return hasMessageAvailableAsync().get();
+} catch (ExecutionException | InterruptedException e) {
+throw new PulsarClientException(e);
+}
+}
+
+public CompletableFuture hasMessageAvailableAsync() {
+final CompletableFuture booleanFuture = new 
CompletableFuture<>();
+
+if (lastMessageIdInBroker.compareTo(lastDequeuedMessage) > 0 &&
+((MessageIdImpl)lastMessageIdInBroker).getEntryId() != -1) {
+booleanFuture.complete(true);
+} else {
+getLastMessageIdAsync().thenAccept(messageId -> {
+lastMessageIdInBroker = messageId;
+if (lastMessageIdInBroker.compareTo(lastDequeuedMessage) > 0 &&
+((MessageIdImpl)lastMessageIdInBroker).getEntryId() != -1) 
{
+booleanFuture.complete(true);
+} else {
+booleanFuture.complete(false);
+}
+}).exceptionally(e -> {
+log.error("[{}][{}] Failed getLastMessageId command", topic, 
subscription);
+booleanFuture.completeExceptionally(e.getCause());
+return null;
+});
+}
+return booleanFuture;
+}
+
+private CompletableFuture getLastMessageIdAsync() {
+if (getState() == State.Closing || getState() == State.Closed) {
+return FutureUtil
+.failedFuture(new 
PulsarClientException.AlreadyClosedException("Consumer was already closed"));
+}
+
+if (!isConnected()) {
+long opTimeoutMs = 
client.getConfiguration().getOperationTimeoutMs();
 
 Review comment:
   Thanks, will change this.


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] zhaijack commented on a change in pull request #1066: Issue 937: add CommandGetLastMessageId to make reader know the end of topic

[GitBox]

zhaijack commented on a change in pull request #1066: Issue 937: add 
CommandGetLastMessageId to make reader know the end of topic
URL: https://github.com/apache/incubator-pulsar/pull/1066#discussion_r167811273
 
 

 ##
 File path: 
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
 ##
 @@ -1248,6 +1254,77 @@ public void seek(MessageId messageId) throws 
PulsarClientException {
 return seekFuture;
 }
 
+public boolean hasMessageAvailable() throws PulsarClientException {
+try {
+return hasMessageAvailableAsync().get();
+} catch (ExecutionException | InterruptedException e) {
+throw new PulsarClientException(e);
+}
+}
+
+public CompletableFuture hasMessageAvailableAsync() {
+final CompletableFuture booleanFuture = new 
CompletableFuture<>();
+
+if (lastMessageIdInBroker.compareTo(lastDequeuedMessage) > 0 &&
 
 Review comment:
   Thanks. will add it.


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] zhaijack commented on a change in pull request #1066: Issue 937: add CommandGetLastMessageId to make reader know the end of topic

[GitBox]

zhaijack commented on a change in pull request #1066: Issue 937: add 
CommandGetLastMessageId to make reader know the end of topic
URL: https://github.com/apache/incubator-pulsar/pull/1066#discussion_r167457406
 
 

 ##
 File path: 
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
 ##
 @@ -1248,6 +1254,77 @@ public void seek(MessageId messageId) throws 
PulsarClientException {
 return seekFuture;
 }
 
+public boolean hasMessageAvailable() throws PulsarClientException {
+try {
+return hasMessageAvailableAsync().get();
+} catch (ExecutionException | InterruptedException e) {
+throw new PulsarClientException(e);
+}
+}
+
+public CompletableFuture hasMessageAvailableAsync() {
+final CompletableFuture booleanFuture = new 
CompletableFuture<>();
+
+if (lastMessageIdInBroker.compareTo(lastDequeuedMessage) > 0 &&
+((MessageIdImpl)lastMessageIdInBroker).getEntryId() != -1) {
+booleanFuture.complete(true);
+} else {
+getLastMessageIdAsync().thenAccept(messageId -> {
+lastMessageIdInBroker = messageId;
+if (lastMessageIdInBroker.compareTo(lastDequeuedMessage) > 0 &&
+((MessageIdImpl)lastMessageIdInBroker).getEntryId() != -1) 
{
+booleanFuture.complete(true);
+} else {
+booleanFuture.complete(false);
+}
+}).exceptionally(e -> {
+log.error("[{}][{}] Failed getLastMessageId command", topic, 
subscription);
+booleanFuture.completeExceptionally(e.getCause());
+return null;
+});
+}
+return booleanFuture;
+}
+
+private CompletableFuture getLastMessageIdAsync() {
+if (getState() == State.Closing || getState() == State.Closed) {
+return FutureUtil
+.failedFuture(new 
PulsarClientException.AlreadyClosedException("Consumer was already closed"));
+}
+
+if (cnx().getRemoteEndpointProtocolVersion() < 
ProtocolVersion.v11.getNumber()) {
 
 Review comment:
   Thanks, will move the statement of isConnected up.


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] zhaijack commented on a change in pull request #1066: Issue 937: add CommandGetLastMessageId to make reader know the end of topic

[GitBox]

zhaijack commented on a change in pull request #1066: Issue 937: add 
CommandGetLastMessageId to make reader know the end of topic
URL: https://github.com/apache/incubator-pulsar/pull/1066#discussion_r167251458
 
 

 ##
 File path: 
pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicReaderTest.java
 ##
 @@ -359,4 +362,59 @@ public EncryptionKeyInfo getPrivateKey(String keyName, 
Map keyMe
 reader.close();
 log.info("-- Exiting {} test --", methodName);
 }
+
+
+@Test
+public void testSimpleReaderReachEndofTopic() throws Exception {
 
 Review comment:
   thanks, will add the test.


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] zhaijack commented on a change in pull request #1066: Issue 937: add CommandGetLastMessageId to make reader know the end of topic

[GitBox]

zhaijack commented on a change in pull request #1066: Issue 937: add 
CommandGetLastMessageId to make reader know the end of topic
URL: https://github.com/apache/incubator-pulsar/pull/1066#discussion_r167156439
 
 

 ##
 File path: pulsar-client/src/main/java/org/apache/pulsar/client/api/Reader.java
 ##
 @@ -62,4 +62,14 @@
  * Return true if the topic was terminated and this reader has reached the 
end of the topic
  */
 boolean hasReachedEndOfTopic();
+
+/**
+ * Check if there is message that has been published successfully to the 
broker in the topic.
+ */
+Boolean hasMessageAvailable() throws PulsarClientException;
 
 Review comment:
   Yes. we have keep the value as `lastMessageIdInBroker` in `ConsumerImpl`, it 
will return from client most of the time, will not call into broker.


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] zhaijack commented on a change in pull request #1066: Issue 937: add CommandGetLastMessageId to make reader know the end of topic

[GitBox]

zhaijack commented on a change in pull request #1066: Issue 937: add 
CommandGetLastMessageId to make reader know the end of topic
URL: https://github.com/apache/incubator-pulsar/pull/1066#discussion_r167153269
 
 

 ##
 File path: 
pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageIdImpl.java
 ##
 @@ -39,10 +39,11 @@
 protected final long entryId;
 protected final int partitionIndex;
 
-// Private constructor used only for json deserialization
-@SuppressWarnings("unused")
-private MessageIdImpl() {
-this(-1, -1, -1);
+// create MessageIdImpl use value of MessageId.earliest
+public MessageIdImpl() {
 
 Review comment:
   Thanks, will change it back, and reference earliest when use this. I thought 
it may make this clearer than hard coded -1,  


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] zhaijack commented on a change in pull request #1066: Issue 937: add CommandGetLastMessageId to make reader know the end of topic

[GitBox]

zhaijack commented on a change in pull request #1066: Issue 937: add 
CommandGetLastMessageId to make reader know the end of topic
URL: https://github.com/apache/incubator-pulsar/pull/1066#discussion_r167148272
 
 

 ##
 File path: 
pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicReaderTest.java
 ##
 @@ -28,8 +31,10 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
-
+import javax.validation.constraints.AssertFalse;
 
 Review comment:
   Thanks. will change it.


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] zhaijack commented on a change in pull request #1066: Issue 937: add CommandGetLastMessageId to make reader know the end of topic

[GitBox]

zhaijack commented on a change in pull request #1066: Issue 937: add 
CommandGetLastMessageId to make reader know the end of topic
URL: https://github.com/apache/incubator-pulsar/pull/1066#discussion_r167148083
 
 

 ##
 File path: 
pulsar-client/src/main/java/org/apache/pulsar/client/api/Consumer.java
 ##
 @@ -278,4 +277,18 @@
  * @return a future to track the completion of the seek operation
  */
 CompletableFuture seekAsync(MessageId messageId);
+
+/**
+ * Check if there is message that has been published successfully to the 
broker in the topic.
+ *
+ * Note: this operation can only be done on non-partitioned persistent 
topics.
+ * For partitioned topics, one can rather perform the hasMessageAvailable 
on the individual partitions.
+ */
+Boolean hasMessageAvailable() throws PulsarClientException;
+
+/**
+ * Asynchronously Check if there is message that has been published 
successfully to the broker in the topic.
+ */
+CompletableFuture hasMessageAvailableAsync();
 
 Review comment:
   Get it. will hide it here.


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] zhaijack commented on a change in pull request #1066: Issue 937: add CommandGetLastMessageId to make reader know the end of topic

[GitBox]

zhaijack commented on a change in pull request #1066: Issue 937: add 
CommandGetLastMessageId to make reader know the end of topic
URL: https://github.com/apache/incubator-pulsar/pull/1066#discussion_r167148039
 
 

 ##
 File path: pulsar-common/src/main/proto/PulsarApi.proto
 ##
 @@ -415,6 +415,16 @@ message CommandConsumerStatsResponse {
 optional uint64 msgBacklog  = 15;
 }
 
+message CommandGetLastMessageId {
 
 Review comment:
   Thanks. will add the checking


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] zhaijack commented on a change in pull request #1066: Issue 937: add CommandGetLastMessageId to make reader know the end of topic

[GitBox]

zhaijack commented on a change in pull request #1066: Issue 937: add 
CommandGetLastMessageId to make reader know the end of topic
URL: https://github.com/apache/incubator-pulsar/pull/1066#discussion_r165329904
 
 

 ##
 File path: pulsar-common/src/main/proto/PulsarApi.proto
 ##
 @@ -415,6 +415,16 @@ message CommandConsumerStatsResponse {
 optional uint64 msgBacklog  = 15;
 }
 
+message CommandGetLastMessageId {
 
 Review comment:
   Thanks, will add it.


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] zhaijack commented on a change in pull request #1066: Issue 937: add CommandGetLastMessageId to make reader know the end of topic

[GitBox]

zhaijack commented on a change in pull request #1066: Issue 937: add 
CommandGetLastMessageId to make reader know the end of topic
URL: https://github.com/apache/incubator-pulsar/pull/1066#discussion_r162976036
 
 

 ##
 File path: 
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
 ##
 @@ -851,6 +849,31 @@ protected void handleCloseConsumer(CommandCloseConsumer 
closeConsumer) {
 }
 }
 
+@Override
+protected void handleGetLastMessageId(CommandGetLastMessageId 
getLastMessageId) {
+checkArgument(state == State.Connected);
+
+CompletableFuture consumerFuture = 
consumers.get(getLastMessageId.getConsumerId());
+
+if (consumerFuture != null && consumerFuture.isDone() && 
!consumerFuture.isCompletedExceptionally()) {
+Consumer consumer = consumerFuture.getNow(null);
+long requestId = getLastMessageId.getRequestId();
+
+Position position = 
consumer.getSubscription().getTopic().getLastMessageId();
+
+log.info("[{}] [{}][{}] Get LastMessageId {}", remoteAddress,
+consumer.getSubscription().getTopic().getName(), 
consumer.getSubscription().getName(), position);
+MessageIdData messageId = MessageIdData.newBuilder()
 
 Review comment:
   noticed this just now, will add partition here, if it is needed.


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] zhaijack commented on a change in pull request #1066: Issue 937: add CommandGetLastMessageId to make reader know the end of topic

[GitBox]

zhaijack commented on a change in pull request #1066: Issue 937: add 
CommandGetLastMessageId to make reader know the end of topic
URL: https://github.com/apache/incubator-pulsar/pull/1066#discussion_r162941099
 
 

 ##
 File path: 
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
 ##
 @@ -851,6 +849,31 @@ protected void handleCloseConsumer(CommandCloseConsumer 
closeConsumer) {
 }
 }
 
+@Override
+protected void handleGetLastMessageId(CommandGetLastMessageId 
getLastMessageId) {
+checkArgument(state == State.Connected);
+
+CompletableFuture consumerFuture = 
consumers.get(getLastMessageId.getConsumerId());
+
+if (consumerFuture != null && consumerFuture.isDone() && 
!consumerFuture.isCompletedExceptionally()) {
+Consumer consumer = consumerFuture.getNow(null);
+long requestId = getLastMessageId.getRequestId();
+
+Position position = 
consumer.getSubscription().getTopic().getLastMessageId();
+
+log.info("[{}] [{}][{}] Get LastMessageId {}", remoteAddress,
+consumer.getSubscription().getTopic().getName(), 
consumer.getSubscription().getName(), position);
+MessageIdData messageId = MessageIdData.newBuilder()
 
 Review comment:
   Thanks. Seems we got here only ledgerid and messageid. will noted in the 
client api that it only support non-partitioned topic.  partition and batch 
index here should be default value.


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] zhaijack commented on a change in pull request #1066: Issue 937: add CommandGetLastMessageId to make reader know the end of topic

[GitBox]

zhaijack commented on a change in pull request #1066: Issue 937: add 
CommandGetLastMessageId to make reader know the end of topic
URL: https://github.com/apache/incubator-pulsar/pull/1066#discussion_r162845520
 
 

 ##
 File path: 
pulsar-client/src/main/java/org/apache/pulsar/client/api/Consumer.java
 ##
 @@ -278,4 +277,19 @@
  * @return a future to track the completion of the seek operation
  */
 CompletableFuture seekAsync(MessageId messageId);
+
+/**
+ * Gets last message id of Topic.
+ *
+ * @return the last message id
+ */
+MessageId getLastMessageId() throws PulsarClientException;
 
 Review comment:
   Thanks for all the comments. 
   If needed, will add a position var to track the last position 
in`incomingMessages` to achieve `consumer.hasMessageAvaiable()`.


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] zhaijack commented on a change in pull request #1066: Issue 937: add CommandGetLastMessageId to make reader know the end of topic

[GitBox]

zhaijack commented on a change in pull request #1066: Issue 937: add 
CommandGetLastMessageId to make reader know the end of topic
URL: https://github.com/apache/incubator-pulsar/pull/1066#discussion_r162833852
 
 

 ##
 File path: pulsar-client/src/main/java/org/apache/pulsar/client/api/Reader.java
 ##
 @@ -62,4 +62,6 @@
  * Return true if the topic was terminated and this reader has reached the 
end of the topic
  */
 boolean hasReachedEndOfTopic();
+
+MessageId getLastMessageId() throws PulsarClientException;
 
 Review comment:
   thanks, will add it.


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] zhaijack commented on a change in pull request #1066: Issue 937: add CommandGetLastMessageId to make reader know the end of topic

[GitBox]

zhaijack commented on a change in pull request #1066: Issue 937: add 
CommandGetLastMessageId to make reader know the end of topic
URL: https://github.com/apache/incubator-pulsar/pull/1066#discussion_r162833840
 
 

 ##
 File path: 
pulsar-client/src/main/java/org/apache/pulsar/client/api/Consumer.java
 ##
 @@ -278,4 +277,19 @@
  * @return a future to track the completion of the seek operation
  */
 CompletableFuture seekAsync(MessageId messageId);
+
+/**
+ * Gets last message id of Topic.
 
 Review comment:
   Thanks. will add it.


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] zhaijack commented on a change in pull request #1066: Issue 937: add CommandGetLastMessageId to make reader know the end of topic

[GitBox]

zhaijack commented on a change in pull request #1066: Issue 937: add 
CommandGetLastMessageId to make reader know the end of topic
URL: https://github.com/apache/incubator-pulsar/pull/1066#discussion_r162799848
 
 

 ##
 File path: 
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
 ##
 @@ -903,6 +904,11 @@ public boolean isEncryptionRequired() {
 return CompletableFuture.completedFuture(null);
 }
 
+@Override
+public Position getLastMessageId() {
+throw new UnsupportedOperationException("getLastMessageId is not 
supported on non-persistent topic");
 
 Review comment:
   Thanks. will change it.


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] zhaijack commented on a change in pull request #1066: Issue 937: add CommandGetLastMessageId to make reader know the end of topic

[GitBox]

zhaijack commented on a change in pull request #1066: Issue 937: add 
CommandGetLastMessageId to make reader know the end of topic
URL: https://github.com/apache/incubator-pulsar/pull/1066#discussion_r162799830
 
 

 ##
 File path: 
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
 ##
 @@ -851,6 +849,31 @@ protected void handleCloseConsumer(CommandCloseConsumer 
closeConsumer) {
 }
 }
 
+@Override
+protected void handleGetLastMessageId(CommandGetLastMessageId 
getLastMessageId) {
+checkArgument(state == State.Connected);
+
+CompletableFuture consumerFuture = 
consumers.get(getLastMessageId.getConsumerId());
+
+if (consumerFuture != null && consumerFuture.isDone() && 
!consumerFuture.isCompletedExceptionally()) {
+Consumer consumer = consumerFuture.getNow(null);
+long requestId = getLastMessageId.getRequestId();
+
+Position position = 
consumer.getSubscription().getTopic().getLastMessageId();
+
+log.info("[{}] [{}][{}] Get LastMessageId {}", remoteAddress,
 
 Review comment:
   Thanks, will change into debug


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] zhaijack commented on a change in pull request #1066: Issue 937: add CommandGetLastMessageId to make reader know the end of topic

[GitBox]

zhaijack commented on a change in pull request #1066: Issue 937: add 
CommandGetLastMessageId to make reader know the end of topic
URL: https://github.com/apache/incubator-pulsar/pull/1066#discussion_r162799826
 
 

 ##
 File path: 
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
 ##
 @@ -323,14 +321,16 @@
 public boolean isTerminated();
 
 /**
- * Returns managed-ledger config 
+ * Returns managed-ledger config
  */
 ManagedLedgerConfig getConfig();
 
 /**
  * Updates managed-ledger config
- * 
+ *
  * @param config
  */
 void setConfig(ManagedLedgerConfig config);
+
+Position getLastConfirmedEntry();
 
 Review comment:
   Thanks, will add if needed


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] zhaijack commented on a change in pull request #1066: Issue 937: add CommandGetLastMessageId to make reader know the end of topic

[GitBox]

zhaijack commented on a change in pull request #1066: Issue 937: add 
CommandGetLastMessageId to make reader know the end of topic
URL: https://github.com/apache/incubator-pulsar/pull/1066#discussion_r162799821
 
 

 ##
 File path: 
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
 ##
 @@ -323,14 +321,16 @@
 public boolean isTerminated();
 
 /**
- * Returns managed-ledger config 
 
 Review comment:
   Thanks, this is the auto changing of intellij-idea for checkstyle.


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services

[GitHub] zhaijack commented on a change in pull request #1066: Issue 937: add CommandGetLastMessageId to make reader know the end of topic

[GitBox]

zhaijack commented on a change in pull request #1066: Issue 937: add 
CommandGetLastMessageId to make reader know the end of topic
URL: https://github.com/apache/incubator-pulsar/pull/1066#discussion_r162799821
 
 

 ##
 File path: 
managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedger.java
 ##
 @@ -323,14 +321,16 @@
 public boolean isTerminated();
 
 /**
- * Returns managed-ledger config 
 
 Review comment:
   Thanks, this is the auto changing of intellij-idea. 


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


With regards,
Apache Git Services