Skip to content

[ClientAPI]Fix hasMessageAvailable() #6362

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sijie merged 12 commits into from
Mar 3, 2020
Merged

[ClientAPI]Fix hasMessageAvailable() #6362

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
c210b31
fix hasMessageAvailable for reader
yjshen Feb 19, 2020
095f932
add test case
yjshen Feb 19, 2020
55fd071
sort id to make test deterministic
yjshen Feb 19, 2020
11102e0
fix comments
yjshen Feb 19, 2020
9b6e600
get batch size async
yjshen Feb 20, 2020
172b638
Merge remote-tracking branch 'apache/master' into has_msg_avalilable
yjshen Feb 24, 2020
2278a9b
move logic to async
yjshen Feb 25, 2020
8f7d4d5
reset lastDequeue to earliest when seek
yjshen Feb 25, 2020
8c895f1
resolve comments
yjshen Feb 25, 2020
3724b44
fix compactor test
yjshen Feb 26, 2020
5af7a12
retrigger
yjshen Feb 26, 2020
c3a94e2
rerun tests
yjshen Feb 26, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
86 changes: 76 additions & 10 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,11 @@
import javax.naming.AuthenticationException;
import javax.net.ssl.SSLSession;

import org.apache.bookkeeper.mledger.AsyncCallbacks;
import org.apache.bookkeeper.mledger.Entry;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.Position;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
import org.apache.bookkeeper.mledger.impl.PositionImpl;
import org.apache.bookkeeper.mledger.util.SafeRun;
import org.apache.commons.lang3.StringUtils;
Expand All @@ -59,6 +63,7 @@
import org.apache.pulsar.broker.service.BrokerServiceException.ServerMetadataException;
import org.apache.pulsar.broker.service.BrokerServiceException.ServiceUnitNotReadyException;
import org.apache.pulsar.broker.service.BrokerServiceException.TopicNotFoundException;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.broker.service.schema.exceptions.IncompatibleSchemaException;
import org.apache.pulsar.broker.service.schema.SchemaRegistryService;
import org.apache.pulsar.broker.web.RestException;
Expand Down Expand Up @@ -1396,22 +1401,83 @@ protected void handleGetLastMessageId(CommandGetLastMessageId getLastMessageId)
Topic topic = consumer.getSubscription().getTopic();
Position position = topic.getLastMessageId();
int partitionIndex = TopicName.getPartitionIndex(topic.getName());
if (log.isDebugEnabled()) {
log.debug("[{}] [{}][{}] Get LastMessageId {} partitionIndex {}", remoteAddress,
topic.getName(), consumer.getSubscription().getName(), position, partitionIndex);
}
MessageIdData messageId = MessageIdData.newBuilder()
.setLedgerId(((PositionImpl)position).getLedgerId())
.setEntryId(((PositionImpl)position).getEntryId())
.setPartition(partitionIndex)
.build();

ctx.writeAndFlush(Commands.newGetLastMessageIdResponse(requestId, messageId));
getLargestBatchIndexWhenPossible(
topic,
(PositionImpl) position,
partitionIndex,
requestId,
consumer.getSubscription().getName());

} else {
ctx.writeAndFlush(Commands.newError(getLastMessageId.getRequestId(), ServerError.MetadataError, "Consumer not found"));
}
}

private void getLargestBatchIndexWhenPossible(
Topic topic,
PositionImpl position,
int partitionIndex,
long requestId,
String subscriptionName) {

PersistentTopic persistentTopic = (PersistentTopic) topic;
ManagedLedgerImpl ml = (ManagedLedgerImpl) persistentTopic.getManagedLedger();

// If it's not pointing to a valid entry, respond messageId of the current position.
if (position.getEntryId() == -1) {
MessageIdData messageId = MessageIdData.newBuilder()
.setLedgerId(position.getLedgerId())
.setEntryId(position.getEntryId())
.setPartition(partitionIndex).build();

ctx.writeAndFlush(Commands.newGetLastMessageIdResponse(requestId, messageId));
}

// For a valid position, we read the entry out and parse the batch size from its metadata.
CompletableFuture<Entry> entryFuture = new CompletableFuture<>();
ml.asyncReadEntry(position, new AsyncCallbacks.ReadEntryCallback() {

@merlimat merlimat Mar 2, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ideally we should be avoiding this read by just recording the number of messages in the last batch, along with the current position.

@Override
public void readEntryComplete(Entry entry, Object ctx) {
entryFuture.complete(entry);
}

@Override
public void readEntryFailed(ManagedLedgerException exception, Object ctx) {
entryFuture.completeExceptionally(exception);
}
}, null);

CompletableFuture<Integer> batchSizeFuture = entryFuture.thenApply(entry -> {
MessageMetadata metadata = Commands.parseMessageMetadata(entry.getDataBuffer());
int batchSize = metadata.getNumMessagesInBatch();
entry.release();
return batchSize;
});

batchSizeFuture.whenComplete((batchSize, e) -> {
if (e != null) {
ctx.writeAndFlush(Commands.newError(
requestId, ServerError.MetadataError, "Failed to get batch size for entry " + e.getMessage()));
} else {
int largestBatchIndex = batchSize > 1 ? batchSize - 1 : -1;

if (log.isDebugEnabled()) {
log.debug("[{}] [{}][{}] Get LastMessageId {} partitionIndex {}", remoteAddress,
topic.getName(), subscriptionName, position, partitionIndex);
}

MessageIdData messageId = MessageIdData.newBuilder()
.setLedgerId(position.getLedgerId())
.setEntryId(position.getEntryId())
.setPartition(partitionIndex)
.setBatchIndex(largestBatchIndex).build();

ctx.writeAndFlush(Commands.newGetLastMessageIdResponse(requestId, messageId));
}
});
}

@Override
protected void handleGetTopicsOfNamespace(CommandGetTopicsOfNamespace commandGetTopicsOfNamespace) {
final long requestId = commandGetTopicsOfNamespace.getRequestId();
Expand Down
6 changes: 5 additions & 1 deletion pulsar-broker/src/main/java/org/apache/pulsar/compaction/TwoPhaseCompactor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,6 +43,7 @@
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.RawMessage;
import org.apache.pulsar.client.api.RawReader;
import org.apache.pulsar.client.impl.MessageIdImpl;
import org.apache.pulsar.client.impl.RawBatchConverter;
import org.apache.pulsar.common.protocol.Commands;
import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
Expand Down Expand Up @@ -95,7 +96,10 @@ private CompletableFuture<PhaseOneResult> phaseOne(RawReader reader) {
} else {
log.info("Commencing phase one of compaction for {}, reading to {}",
reader.getTopic(), lastMessageId);
phaseOneLoop(reader, Optional.empty(), Optional.empty(), lastMessageId, latestForKey,
// Each entry is processed as a whole, discard the batchIndex part deliberately.
MessageIdImpl lastImpl = (MessageIdImpl) lastMessageId;
MessageIdImpl lastEntryMessageId = new MessageIdImpl(lastImpl.getLedgerId(), lastImpl.getEntryId(), lastImpl.getPartitionIndex());
phaseOneLoop(reader, Optional.empty(), Optional.empty(), lastEntryMessageId, latestForKey,
loopPromise);
}
});
Expand Down
75 changes: 74 additions & 1 deletion pulsar-broker/src/test/java/org/apache/pulsar/client/api/TopicReaderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,17 @@ public static Object[][] variationsForResetOnLatestMsg() {
};
}

@DataProvider
public static Object[][] variationsForHasMessageAvailable() {
return new Object[][] {
// batching / start-inclusive
{true, true},
{true, false},
{false, true},
{false, false},
};
}

@Test
public void testSimpleReader() throws Exception {
Reader<byte[]> reader = pulsarClient.newReader().topic("persistent://my-property/my-ns/testSimpleReader")
Expand Down Expand Up @@ -531,6 +542,68 @@ public void testMessageAvailableAfterRestart() throws Exception {

}

@Test(dataProvider = "variationsForHasMessageAvailable")
public void testHasMessageAvailable(boolean enableBatch, boolean startInclusive) throws Exception {
final String topicName = "persistent://my-property/my-ns/HasMessageAvailable";
final int numOfMessage = 100;

ProducerBuilder<byte[]> producerBuilder = pulsarClient.newProducer()
.topic(topicName);

if (enableBatch) {
producerBuilder
.enableBatching(true)
.batchingMaxMessages(10);
} else {
producerBuilder
.enableBatching(false);
}

Producer<byte[]> producer = producerBuilder.create();

CountDownLatch latch = new CountDownLatch(numOfMessage);

List<MessageId> allIds = Collections.synchronizedList(new ArrayList<>());

for (int i = 0; i < numOfMessage; i++) {
producer.sendAsync(String.format("msg num %d", i).getBytes()).whenComplete((mid, e) -> {
if (e != null) {
Assert.fail();
} else {
allIds.add(mid);
}
latch.countDown();
});
}

latch.await();

allIds.sort(null); // make sure the largest mid appears at last.

for (MessageId id : allIds) {
Reader<byte[]> reader;

if (startInclusive) {
reader = pulsarClient.newReader().topic(topicName)
.startMessageId(id).startMessageIdInclusive().create();
} else {
reader = pulsarClient.newReader().topic(topicName)
.startMessageId(id).create();
}

if (startInclusive) {
assertTrue(reader.hasMessageAvailable());
} else if (id != allIds.get(allIds.size() - 1)) {
assertTrue(reader.hasMessageAvailable());
} else {
assertFalse(reader.hasMessageAvailable());
}
reader.close();
}

producer.close();
}

@Test
public void testReaderNonDurableIsAbleToSeekRelativeTime() throws Exception {
final int numOfMessage = 10;
Expand Down Expand Up @@ -794,7 +867,7 @@ public void testReaderStartInMiddleOfBatch() throws Exception {
.batchingMaxMessages(10)
.create();

CountDownLatch latch = new CountDownLatch(100);
CountDownLatch latch = new CountDownLatch(numOfMessage);

List<MessageId> allIds = Collections.synchronizedList(new ArrayList<>());

Expand Down
4 changes: 3 additions & 1 deletion pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ReaderTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,7 +145,9 @@ public void testReadMessageWithBatchingWithMessageInclusive() throws Exception {
while (reader.hasMessageAvailable()) {
Assert.assertTrue(keys.remove(reader.readNext().getKey()));
}
Assert.assertTrue(keys.isEmpty());
// start from latest with start message inclusive should only read the last message in batch
Assert.assertTrue(keys.size() == 9);
Assert.assertFalse(keys.contains("key9"));
Assert.assertFalse(reader.hasMessageAvailable());
}

Expand Down
Loading