#2059

Committed 15 Jul 2025 06:40PM UTC coverage: 95.593% (-0.06%) from 95.655%

Build # #2059

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1357 from nats-io/shutdown-internal-connection-executors

[FIX] Shutdown internal executors on connection close.

Run Details

19 of 19 new or added lines in 2 files covered. (100.0%)

16 existing lines in 7 files now uncovered.

11844 of 12390 relevant lines covered (95.59%)

0.96 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.5

/src/main/java/io/nats/client/impl/OrderedMessageManager.java

// Copyright 2021 The NATS Authors
// Licensed 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 io.nats.client.impl;

import io.nats.client.Message;
import io.nats.client.SubscribeOptions;
import io.nats.client.api.ConsumerConfiguration;
import io.nats.client.api.ConsumerCreateRequest;
import io.nats.client.api.ConsumerInfo;

import java.util.concurrent.atomic.AtomicReference;

import static io.nats.client.impl.MessageManager.ManageResult.MESSAGE;
import static io.nats.client.impl.MessageManager.ManageResult.STATUS_HANDLED;
import static io.nats.client.support.NatsJetStreamUtil.generateConsumerName;

class OrderedMessageManager extends PushMessageManager {

    protected long expectedExternalConsumerSeq;
    protected final AtomicReference<String> targetSid;

    protected OrderedMessageManager(
        NatsConnection conn,
        NatsJetStream js,
        String stream,
        SubscribeOptions so,
        ConsumerConfiguration originalCc,
        boolean queueMode,
        boolean syncMode)
    {
        super(conn, js, stream, so, originalCc, queueMode, syncMode);
        expectedExternalConsumerSeq = 1; // always starts at 1
        targetSid = new AtomicReference<>();
    }

    @Override
    protected void startup(NatsJetStreamSubscription sub) {
        expectedExternalConsumerSeq = 1; // consumer always starts with consumer sequence 1
        super.startup(sub);
        targetSid.set(sub.getSID());
    }

    @Override
    protected ManageResult manage(Message msg) {
        if (!msg.getSID().equals(targetSid.get())) {
            return STATUS_HANDLED; // wrong sid. message is a throwaway from previous consumer that errored
        }

        if (msg.isJetStream()) {
            long receivedConsumerSeq = msg.metaData().consumerSequence();
            if (expectedExternalConsumerSeq != receivedConsumerSeq) {
                handleErrorCondition();
                return STATUS_HANDLED;
            }
            trackJsMessage(msg);
            expectedExternalConsumerSeq++;
            return MESSAGE;
        }

        return manageStatus(msg);
    }

    @Override
    protected void handleHeartbeatError() {
        super.handleHeartbeatError();
        handleErrorCondition();
    }

    private void handleErrorCondition() {
        try {
            targetSid.set(null);
            expectedExternalConsumerSeq = 1; // consumer always starts with consumer sequence 1

            // 1. re-subscribe. This means killing the sub then making a new one.
            //    New sub needs a new deliverSubject
            String newDeliverSubject = sub.connection.createInbox();
            sub.reSubscribe(newDeliverSubject);
            targetSid.set(sub.getSID());

            // 2a. make a new consumer using the same "deliver" subject but with a new starting point, and a new name
            ConsumerConfiguration.Builder b = js.consumerConfigurationForOrdered(initialCc, lastStreamSeq, newDeliverSubject, null);

            // 2b. because we bypass the normal create-subscription workflow,
            //     we have to handle the fact that ordered consumers must always have a unique name.
            //     if the user supplied a name, well call generateConsumerName with the original name as a prefix
            if (initialCc.getName() != null) {
                b.name(generateConsumerName(initialCc.getName()));
            }
            ConsumerConfiguration userCC = b.build();
            ConsumerInfo ci = js._createConsumer(stream, userCC, ConsumerCreateRequest.Action.Create); // this can fail when a server is down.
            sub.setConsumerName(ci.getName());

            // 3. restart the manager.
            startup(sub);
        }
        catch (Exception e) {
            // don't want this doubly failing for any reason
            try {
                js.conn.processException(e);
            }
            catch (Exception ignore) {}
            initOrResetHeartbeatTimer();
        }
    }
}

1	// Copyright 2021 The NATS Authors
2	// Licensed under the Apache License, Version 2.0 (the "License");
3	// you may not use this file except in compliance with the License.
4	// You may obtain a copy of the License at:
5	//
6	// http://www.apache.org/licenses/LICENSE-2.0
7	//
8	// Unless required by applicable law or agreed to in writing, software
9	// distributed under the License is distributed on an "AS IS" BASIS,
10	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11	// See the License for the specific language governing permissions and
12	// limitations under the License.
13
14	package io.nats.client.impl;
15
16	import io.nats.client.Message;
17	import io.nats.client.SubscribeOptions;
18	import io.nats.client.api.ConsumerConfiguration;
19	import io.nats.client.api.ConsumerCreateRequest;
20	import io.nats.client.api.ConsumerInfo;
21
22	import java.util.concurrent.atomic.AtomicReference;
23
24	import static io.nats.client.impl.MessageManager.ManageResult.MESSAGE;
25	import static io.nats.client.impl.MessageManager.ManageResult.STATUS_HANDLED;
26	import static io.nats.client.support.NatsJetStreamUtil.generateConsumerName;
27
28	class OrderedMessageManager extends PushMessageManager {
29
30	protected long expectedExternalConsumerSeq;
31	protected final AtomicReference<String> targetSid;
32
33	protected OrderedMessageManager(
34	NatsConnection conn,
35	NatsJetStream js,
36	String stream,
37	SubscribeOptions so,
38	ConsumerConfiguration originalCc,
39	boolean queueMode,
40	boolean syncMode)
41	{
42	super(conn, js, stream, so, originalCc, queueMode, syncMode);	1✔
43	expectedExternalConsumerSeq = 1; // always starts at 1	1✔
44	targetSid = new AtomicReference<>();	1✔
45	}	1✔
46
47	@Override
48	protected void startup(NatsJetStreamSubscription sub) {
49	expectedExternalConsumerSeq = 1; // consumer always starts with consumer sequence 1	1✔
50	super.startup(sub);	1✔
51	targetSid.set(sub.getSID());	1✔
52	}	1✔
53
54	@Override
55	protected ManageResult manage(Message msg) {
56	if (!msg.getSID().equals(targetSid.get())) {	1✔
57	return STATUS_HANDLED; // wrong sid. message is a throwaway from previous consumer that errored	1✔
58	}
59
60	if (msg.isJetStream()) {	1✔
61	long receivedConsumerSeq = msg.metaData().consumerSequence();	1✔
62	if (expectedExternalConsumerSeq != receivedConsumerSeq) {	1✔
63	handleErrorCondition();	1✔
64	return STATUS_HANDLED;	1✔
65	}
66	trackJsMessage(msg);	1✔
67	expectedExternalConsumerSeq++;	1✔
68	return MESSAGE;	1✔
69	}
70
71	return manageStatus(msg);	×
72	}
73
74	@Override
75	protected void handleHeartbeatError() {
76	super.handleHeartbeatError();	1✔
77	handleErrorCondition();	1✔
78	}	1✔
79
80	private void handleErrorCondition() {
81	try {
82	targetSid.set(null);	1✔
83	expectedExternalConsumerSeq = 1; // consumer always starts with consumer sequence 1	1✔
84
85	// 1. re-subscribe. This means killing the sub then making a new one.
86	// New sub needs a new deliverSubject
87	String newDeliverSubject = sub.connection.createInbox();	1✔
88	sub.reSubscribe(newDeliverSubject);	1✔
89	targetSid.set(sub.getSID());	1✔
90
91	// 2a. make a new consumer using the same "deliver" subject but with a new starting point, and a new name
92	ConsumerConfiguration.Builder b = js.consumerConfigurationForOrdered(initialCc, lastStreamSeq, newDeliverSubject, null);	1✔
93
94	// 2b. because we bypass the normal create-subscription workflow,
95	// we have to handle the fact that ordered consumers must always have a unique name.
96	// if the user supplied a name, well call generateConsumerName with the original name as a prefix
97	if (initialCc.getName() != null) {	1✔
98	b.name(generateConsumerName(initialCc.getName()));	1✔
99	}
100	ConsumerConfiguration userCC = b.build();	1✔
101	ConsumerInfo ci = js._createConsumer(stream, userCC, ConsumerCreateRequest.Action.Create); // this can fail when a server is down.	1✔
102	sub.setConsumerName(ci.getName());	1✔
103
104	// 3. restart the manager.
105	startup(sub);	1✔
106	}
UNCOV 107	catch (Exception e) {	×
108	// don't want this doubly failing for any reason
109	try {
110	js.conn.processException(e);	×
111	}
UNCOV 112	catch (Exception ignore) {}	×
UNCOV 113	initOrResetHeartbeatTimer();	×
114	}	1✔
115	}	1✔
116	}

nats-io / nats.java / #2059

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous