pulsar-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From si...@apache.org
Subject [pulsar] branch master updated: support GetLastMessageId api (#3196)
Date Fri, 21 Dec 2018 05:19:33 GMT
This is an automated email from the ASF dual-hosted git repository.

sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new f493676  support GetLastMessageId api (#3196)
f493676 is described below

commit f4936763af1c1dab54d031ae355e40d66d3a602e
Author: legendtkl <taokelu@gmail.com>
AuthorDate: Fri Dec 21 13:19:28 2018 +0800

    support GetLastMessageId api (#3196)
    
    issue ticket:  https://github.com/apache/pulsar/issues/3162 (Fixes #3162)
    
    the pr is straightforward, and it exposes the Persistent Topic GetLastMessageId to Rest
API: /admin/v2/persistent/{tenant}/{namespace}/{topic}/lastMessageId
---
 .../broker/admin/impl/PersistentTopicsBase.java    |  17 +++
 .../pulsar/broker/admin/v2/PersistentTopics.java   |  14 ++
 .../broker/admin/AdminApiGetLastMessageIdTest.java | 142 +++++++++++++++++++++
 3 files changed, 173 insertions(+)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
index 1605d8c..de1af66 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
@@ -1450,4 +1450,21 @@ public class PersistentTopicsBase extends AdminResource {
         }
         return;
     }
+
+    protected MessageId internalGetLastMessageId(boolean authoritative) {
+        validateAdminOperationOnTopic(authoritative);
+
+        if (!(getTopicReference(topicName) instanceof PersistentTopic)) {
+            log.error("[{}] Not supported operation of non-persistent topic {}", clientAppId(),
topicName);
+            throw new RestException(Status.METHOD_NOT_ALLOWED,
+                    "GetLastMessageId on a non-persistent topic is not allowed");
+        }
+        PersistentTopic topic = (PersistentTopic) getTopicReference(topicName);
+        Position position = topic.getLastMessageId();
+        int partitionIndex = TopicName.getPartitionIndex(topic.getName());
+
+        MessageId messageId = new MessageIdImpl(((PositionImpl)position).getLedgerId(), ((PositionImpl)position).getEntryId(),
partitionIndex);
+
+        return messageId;
+    }
 }
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
index 437407b..350da17 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
@@ -498,4 +498,18 @@ public class PersistentTopics extends PersistentTopicsBase {
         validateTopicName(tenant, namespace, encodedTopic);
         return internalOffloadStatus(authoritative);
     }
+
+    @GET
+    @Path("/{tenant}/{namespace}/{topic}/lastMessageId")
+    @ApiOperation(value = "Return the last commit message id of topic")
+    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
+            @ApiResponse(code = 405, message = "Operation not allowed on persistent topic"),
+            @ApiResponse(code = 404, message = "Topic does not exist")})
+    public MessageId getLastMessageId(@PathParam("tenant") String tenant,
+                                                    @PathParam("namespace") String namespace,
+                                                    @PathParam("topic") @Encoded String encodedTopic,
+                                                    @QueryParam("authoritative") @DefaultValue("false")
boolean authoritative) {
+        validateTopicName(tenant, namespace, encodedTopic);
+        return internalGetLastMessageId(authoritative);
+    }
 }
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiGetLastMessageIdTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiGetLastMessageIdTest.java
new file mode 100644
index 0000000..cc5cf0e
--- /dev/null
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiGetLastMessageIdTest.java
@@ -0,0 +1,142 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.admin;
+
+import com.google.common.collect.Sets;
+import org.apache.pulsar.broker.admin.v2.PersistentTopics;
+import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.broker.authentication.AuthenticationDataHttps;
+import org.apache.pulsar.broker.web.PulsarWebResource;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.MessageRoutingMode;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.impl.MessageIdImpl;
+import org.apache.pulsar.common.policies.data.ClusterData;
+import org.apache.pulsar.common.policies.data.TenantInfo;
+import org.testng.Assert;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import javax.ws.rs.core.UriInfo;
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.Future;
+
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+
+public class AdminApiGetLastMessageIdTest extends MockedPulsarServiceBaseTest {
+
+    private PersistentTopics persistentTopics;
+    private final String testTenant = "my-tenant";
+    private final String testLocalCluster = "use";
+    private final String testNamespace = "my-namespace";
+    protected Field uriField;
+    protected UriInfo uriInfo;
+
+    @BeforeClass
+    public void initPersistentTopics() throws Exception {
+        uriField = PulsarWebResource.class.getDeclaredField("uri");
+        uriField.setAccessible(true);
+        uriInfo = mock(UriInfo.class);
+    }
+
+    @Override
+    @BeforeMethod
+    protected void setup() throws Exception {
+        super.internalSetup();
+        admin.clusters().createCluster("test", new ClusterData(brokerUrl.toString()));
+        admin.tenants().createTenant("prop",
+                new TenantInfo(Sets.newHashSet("appid1"), Sets.newHashSet("test")));
+        admin.namespaces().createNamespace("prop/ns-abc");
+        admin.namespaces().setNamespaceReplicationClusters("prop/ns-abc", Sets.newHashSet("test"));
+        persistentTopics = spy(new PersistentTopics());
+        persistentTopics.setServletContext(new MockServletContext());
+        persistentTopics.setPulsar(pulsar);
+
+        doReturn(mockZookKeeper).when(persistentTopics).globalZk();
+        doReturn(mockZookKeeper).when(persistentTopics).localZk();
+        doReturn(pulsar.getConfigurationCache().propertiesCache()).when(persistentTopics).tenantsCache();
+        doReturn(pulsar.getConfigurationCache().policiesCache()).when(persistentTopics).policiesCache();
+        doReturn(false).when(persistentTopics).isRequestHttps();
+        doReturn(null).when(persistentTopics).originalPrincipal();
+        doReturn("test").when(persistentTopics).clientAppId();
+        doReturn("persistent").when(persistentTopics).domain();
+        doNothing().when(persistentTopics).validateAdminAccessForTenant(this.testTenant);
+        doReturn(mock(AuthenticationDataHttps.class)).when(persistentTopics).clientAuthData();
+    }
+
+    @Override
+    @AfterMethod
+    protected void cleanup() throws Exception {
+        super.internalCleanup();
+    }
+
+    @Test
+    public void testGetLastMessageId() throws Exception {
+        try {
+            persistentTopics.getLastMessageId(testTenant, testNamespace, "my-topic", true);
+        } catch (Exception e) {
+            //System.out.println(e.getMessage());
+            Assert.assertEquals("Topic not found", e.getMessage());
+        }
+
+        String key = "legendtkl";
+        final String topicName = "persistent://prop/ns-abc/my-topic";
+        final String messagePredicate = "my-message-" + key + "-";
+        final int numberOfMessages = 30;
+
+        // 2. Create Producer
+        Producer<byte[]> producer = pulsarClient.newProducer().topic(topicName)
+                .enableBatching(false)
+                .messageRoutingMode(MessageRoutingMode.SinglePartition)
+                .create();
+
+        // 3. Publish message and get message id
+        for (int i = 0; i < numberOfMessages; i++) {
+            String message = messagePredicate + i;
+            producer.send(message.getBytes());
+        }
+
+        MessageId id = persistentTopics.getLastMessageId("prop", "ns-abc", "my-topic", true);
+        System.out.println(id.toString());
+        Assert.assertTrue(((MessageIdImpl)id).getLedgerId() >= 0);
+        Assert.assertEquals(numberOfMessages-1, ((MessageIdImpl)id).getEntryId());
+
+        // send more numberOfMessages messages, the last message id should be numberOfMessages*2-1
+        for (int i = 0; i < numberOfMessages; i++) {
+            String message = messagePredicate + i;
+            producer.send(message.getBytes());
+        }
+        id = persistentTopics.getLastMessageId("prop", "ns-abc", "my-topic", true);
+        System.out.println(id.toString());
+        Assert.assertTrue(((MessageIdImpl)id).getLedgerId() > 0);
+        Assert.assertEquals( 2 * numberOfMessages -1, ((MessageIdImpl)id).getEntryId());
+
+        System.out.println(id.toString());
+    }
+}


Mime
View raw message