Skip to content

[BUG] MessageConsumer.isFinished() not set properly in certain conditions #1339

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
merged 2 commits into from
Jun 27, 2025
Merged

[BUG] MessageConsumer.isFinished() not set properly in certain conditions #1339

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
48b41f3
[BUG] MessageConsumer.isFinished() not set properly in certain condit…
scottf Jun 26, 2025
5a5df2c
finishAndClose();
scottf Jun 26, 2025
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
12 changes: 4 additions & 8 deletions src/main/java/io/nats/client/impl/NatsFetchConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,8 +67,7 @@ public void pendingUpdated() {}

@Override
public void heartbeatError() {
stopped.set(true);
finished.set(true);
finishAndClose();
Copy link
Contributor Author

Choose a reason for hiding this comment

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

a heartbeat error on a fetch is terminal

}

@Override
Expand All @@ -86,8 +85,7 @@ public Message nextMessage() throws InterruptedException, JetStreamStatusChecked
if (m == null) {
// if there are no messages in the internal cache AND there are no more pending,
// they all have been read and we can go ahead and finish
finished.set(true);
lenientClose();
finishAndClose();
Copy link
Contributor Author

@scottf scottf Jun 26, 2025 •
edited
Loading

Choose a reason for hiding this comment

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

no more messages on a fetch is terminal

}
return m;
}
Expand All @@ -105,17 +103,15 @@ public Message nextMessage() throws InterruptedException, JetStreamStatusChecked
Message m = sub._nextUnmanagedNoWait(pullSubject);
if (m == null) {
// no message and no time left, go ahead and finish
finished.set(true);
lenientClose();
finishAndClose();
Copy link
Contributor Author

@scottf scottf Jun 26, 2025 •
edited
Loading

Choose a reason for hiding this comment

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

no more messages on a fetch is terminal

}
return m;
}

Message m = sub._nextUnmanaged(timeLeftNanos, pullSubject);
if (m == null && isNoWaitNoExpires) {
// no message and no wait, go ahead and finish
finished.set(true);
lenientClose();
finishAndClose();
Copy link
Contributor Author

@scottf scottf Jun 26, 2025 •
edited
Loading

Choose a reason for hiding this comment

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

no more messages on a fetch is terminal

}
return m;
}
Expand Down
20 changes: 15 additions & 5 deletions src/main/java/io/nats/client/impl/NatsMessageConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,9 +53,14 @@ class NatsMessageConsumer extends NatsMessageConsumerBase implements PullManager
public void heartbeatError() {
try {
// just close the current sub and make another one.
// this could go on endlessly
lenientClose();
doSub();
// this could go on endlessly - unless the user had called stop
if (stopped.get()) {
finishAndClose();
}
else {
Copy link
Contributor Author

Choose a reason for hiding this comment

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

If a heartbeat error occurs, we would usually just try again. But if the user had called stop, they've told us that we are finished.

lenientClose();
doSub();
}
}
catch (JetStreamApiException | IOException e) {
pmm.resetTracking();
Expand All @@ -67,7 +72,7 @@ void doSub() throws JetStreamApiException, IOException {
MessageHandler mh = userMessageHandler == null ? null : msg -> {
userMessageHandler.onMessage(msg);
if (stopped.get() && pmm.noMorePending()) {
finished.set(true);
finishAndClose();
Copy link
Contributor Author

Choose a reason for hiding this comment

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

We have no more pending and the user has called stop, so we are finished with this consume

}
};
super.initSub(subscriptionMaker.subscribe(mh, userDispatcher, pmm, null));
Expand All @@ -78,7 +83,12 @@ void doSub() throws JetStreamApiException, IOException {

@Override
public void pendingUpdated() {
if (!stopped.get() && (pmm.pendingMessages <= thresholdMessages || (pmm.trackingBytes && pmm.pendingBytes <= thresholdBytes)))
if (stopped.get()) {
Copy link
Contributor Author

Choose a reason for hiding this comment

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

We have no more pending and the user has called stop, so we are finished with this consume

if (pmm.noMorePending()) {
finishAndClose();
}
}
else if (pmm.pendingMessages <= thresholdMessages || (pmm.trackingBytes && pmm.pendingBytes <= thresholdBytes))
{
repull();
}
Expand Down
8 changes: 8 additions & 0 deletions src/main/java/io/nats/client/impl/NatsMessageConsumerBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,6 +104,14 @@ public void close() throws Exception {
lenientClose();
}

protected void finishAndClose() {
if (pmm != null) {
pmm.shutdownHeartbeatTimer();
}
finished.set(true);
lenientClose();
}

protected void lenientClose() {
try {
if (!stopped.get() || sub.isActive()) {
Expand Down
2 changes: 1 addition & 1 deletion src/main/java/io/nats/client/impl/PullMessageManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -118,7 +118,7 @@ protected Boolean beforeQueueProcessorImpl(NatsMessage msg) {

// heartbeat just needed to be recorded
if (status.isHeartbeat()) {
trackIncoming(Integer.MIN_VALUE, Integer.MIN_VALUE);
updateLastMessageReceived(); // no need to call track incoming, this is all it does
return false;
}

Expand Down
36 changes: 33 additions & 3 deletions src/test/java/io/nats/client/impl/SimplificationTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1303,20 +1303,20 @@ public void testOverflowFetch() throws Exception {

FetchConsumeOptions fcoNoMin = FetchConsumeOptions.builder()
.maxMessages(5)
.expiresIn(2000)
.expiresIn(3000)
.group(group)
.build();

FetchConsumeOptions fcoOverA = FetchConsumeOptions.builder()
.maxMessages(5)
.expiresIn(2000)
.expiresIn(3000)
.group(group)
.minAckPending(5)
.build();

FetchConsumeOptions fcoOverB = FetchConsumeOptions.builder()
.maxMessages(5)
.expiresIn(2000)
.expiresIn(3000)
Copy link
Contributor Author

Choose a reason for hiding this comment

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

this test was flapping, so added a little extra time

.group(group)
.minAckPending(6)
.build();
Expand Down Expand Up @@ -1495,4 +1495,34 @@ public void testOverflowConsume() throws Exception {
assertEquals(0, overCount.get());
});
}

@Test
public void testFinishEmptyStream() throws Exception {
ListenerForTesting l = new ListenerForTesting();
Options.Builder b = Options.builder().errorListener(l);
runInJsServer(b, nc -> {
JetStreamManagement jsm = nc.jetStreamManagement();
TestingStreamContainer tsc = new TestingStreamContainer(jsm);

String name = variant();

ConsumerConfiguration cc = ConsumerConfiguration.builder()
.name(name)
.filterSubjects(tsc.subject()).build();
jsm.addOrUpdateConsumer(tsc.stream, cc);

ConsumerContext cctx = nc.getConsumerContext(tsc.stream, name);

MessageHandler handler = m -> {
System.out.println(m);
m.ack();
};

ConsumeOptions co = ConsumeOptions.builder().expiresIn(1000).build();
MessageConsumer consumer = cctx.consume(co, handler);
consumer.stop();
sleep(1200); // more than the expires period for the consume
assertTrue(consumer.isFinished());
});
}
}
Loading