|
26 | 26 | import java.util.Optional;
|
27 | 27 | import java.util.Set;
|
28 | 28 | import java.util.concurrent.ExecutionException;
|
| 29 | +import lombok.Cleanup; |
29 | 30 | import lombok.extern.slf4j.Slf4j;
|
30 | 31 | import org.apache.commons.lang3.RandomStringUtils;
|
31 | 32 | import org.apache.pulsar.broker.MultiBrokerBaseTest;
|
32 | 33 | import org.apache.pulsar.broker.PulsarService;
|
33 | 34 | import org.apache.pulsar.broker.loadbalance.LeaderBroker;
|
34 | 35 | import org.apache.pulsar.client.admin.PulsarAdmin;
|
35 | 36 | import org.apache.pulsar.client.admin.PulsarAdminException;
|
| 37 | +import org.apache.pulsar.client.api.MessageId; |
| 38 | +import org.apache.pulsar.client.api.Producer; |
36 | 39 | import org.apache.pulsar.common.naming.TopicDomain;
|
37 | 40 | import org.apache.pulsar.common.naming.TopicName;
|
38 | 41 | import org.apache.pulsar.common.policies.data.AutoTopicCreationOverride;
|
@@ -122,4 +125,27 @@ public void testTopicLookup(TopicDomain topicDomain, boolean isPartition) throws
|
122 | 125 | Assert.assertEquals(lookupResultSet.size(), 1);
|
123 | 126 | }
|
124 | 127 |
|
| 128 | + @Test(timeOut = 30 * 1000) |
| 129 | + public void testTopicGetMessageIdByTimestamp() throws Exception { |
| 130 | + PulsarAdmin admin0 = getAllAdmins().get(0); |
| 131 | + String topic = "public/default/t1"; |
| 132 | + admin0.topics().createPartitionedTopic(topic, 1); |
| 133 | + pulsarClient = newPulsarClient(lookupUrl.toString(), 0); |
| 134 | + @Cleanup |
| 135 | + Producer<byte[]> producer = pulsarClient.newProducer().topic(topic).create(); |
| 136 | + for (int i = 0; i < 20; i++) { |
| 137 | + producer.send("msg".getBytes()); |
| 138 | + } |
| 139 | + String brokerUrl = admin0.lookups().lookupTopic(topic + "-partition-0"); |
| 140 | + PulsarAdmin admin = null; |
| 141 | + for (PulsarService additionalBroker : additionalBrokers) { |
| 142 | + if (!brokerUrl.endsWith(String.valueOf(additionalBroker.getBrokerListenPort().get()))) { |
| 143 | + admin = additionalBroker.getAdminClient(); |
| 144 | + } |
| 145 | + } |
| 146 | + Assert.assertNotNull(admin); |
| 147 | + MessageId msgId = |
| 148 | + admin.topics().getMessageIdByTimestamp(topic + "-partition-0", System.currentTimeMillis()); |
| 149 | + Assert.assertNotNull(msgId); |
| 150 | + } |
125 | 151 | }
|
0 commit comments