36783609042

Committed 30 Sep 2026 10:06PM UTC coverage: 90.857% (-0.01%) from 90.87%

Build # 36783609042

Build Type

push

github

Committed by

scottf

Commit Message

Cache Gradle dependencies in module builds

Parallel module jobs each downloaded every dependency from Maven Central, and run 36782032543 failed kv with 429 Too Many Requests. setup-java caches ~/.gradle keyed on the build files; checkout moves first so the key can hash them.

Coverage Stats

4189 of 4830 branches covered (86.73%)

Branch coverage included in aggregate %.

10648 of 11500 relevant lines covered (92.59%)

4.12 hits per line

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

83.14

/jetstream/src/main/java/io/synadia/client/impl/NatsMessageConsumer.java

package io.synadia.client.impl;

import io.synadia.client.Message;
import io.synadia.client.MessageHandler;
import io.synadia.client.api.ConsumerInfo;
import io.synadia.client.api.JetStreamException;


class NatsMessageConsumer extends NatsMessageConsumerBase implements PullManagerObserver {
    protected final ConsumeOptions consumeOpts;
    protected final SimplifiedSubscriptionMaker subscriptionMaker;
    protected final NatsDispatcher userDispatcher;
    protected final MessageHandler userMessageHandler;

    protected final int thresholdMessages;
    protected final long thresholdBytes;
    protected final boolean isTrackingBytes;

    protected int pendingReceivedMessages;
    protected long pendingReceivedBytes;
    protected boolean noReceivedArePending;
    protected boolean forcePull;
    protected int pendingProcessedMessages;
    protected long pendingProcessedBytes;
    protected boolean processedHasCrossedThreshold;

    NatsMessageConsumer(SimplifiedSubscriptionMaker subscriptionMaker,
                        ConsumerInfo cachedConsumerInfo,
                        ConsumeOptions consumeOpts,
                        NatsDispatcher userDispatcher,
                        final MessageHandler userMessageHandler) throws JetStreamException, InterruptedException
    {
        super(cachedConsumerInfo);

        this.subscriptionMaker = subscriptionMaker;
        this.consumeOpts = consumeOpts;
        this.userDispatcher = userDispatcher;
        this.userMessageHandler = userMessageHandler;

        int bm = consumeOpts.getBatchSize();
        long bb = consumeOpts.getBatchBytes();
        int rePullMessages = Math.max(1, bm * consumeOpts.getThresholdPercent() / 100);
        long rePullBytes = bb == 0 ? 0 : Math.max(1, bb * consumeOpts.getThresholdPercent() / 100);
        // The threshold is what the pending count has to drop below for the next pull. A batch of 1
        // gives a re-pull size of 1 and so a threshold of 0, which pending can never go below, so the
        // consumer pulled once and then stopped. Never let the threshold fall under 1.
        int thresholdM = bm - rePullMessages;
        thresholdMessages = thresholdM < 1 ? 1 : thresholdM;
        long thresholdB = bb - rePullBytes;
        thresholdBytes = bb == 0 ? Integer.MIN_VALUE : (thresholdB < 1 ? 1 : thresholdB);
        isTrackingBytes = rePullBytes > 0;
        doSub(true);
    }

    protected void fullResetPending() {
        pendingReceivedMessages = 0;
        pendingReceivedBytes = 0;
        noReceivedArePending = true;
        forcePull = true;
        pendingProcessedMessages = 0;
        pendingProcessedBytes = 0;
        processedHasCrossedThreshold = true;
    }

    protected void statusAdjustPending(int messages, long bytes) {
        pendingReceivedMessages = Math.max(0, pendingReceivedMessages - messages);
        pendingReceivedBytes = Math.max(0, pendingReceivedBytes - bytes);
        noReceivedArePending = pendingReceivedMessages == 0 || (isTrackingBytes && pendingReceivedBytes == 0);
        forcePull = true;
        pendingProcessedMessages = Math.max(0, pendingProcessedMessages - messages);
        pendingProcessedBytes = Math.max(0, pendingProcessedBytes - bytes);
        processedHasCrossedThreshold = pendingProcessedMessages < thresholdMessages || (isTrackingBytes && pendingProcessedBytes < thresholdBytes);
    }

    protected void aboutToPull(int messages, long bytes) {
        pendingReceivedMessages += messages;
        pendingReceivedBytes += bytes;
        noReceivedArePending = false;
        forcePull = false;
        pendingProcessedMessages += messages;
        pendingProcessedBytes += bytes;
        processedHasCrossedThreshold = false;
    }

    protected void updateProcessed(Message msg) {
        pendingProcessedMessages = Math.max(0, pendingProcessedMessages - 1);
        if (pendingProcessedMessages < thresholdMessages) {
            processedHasCrossedThreshold = true;
        }
        if (isTrackingBytes) {
            pendingProcessedBytes = Math.max(0, pendingProcessedBytes - msg.consumeByteCount());
            processedHasCrossedThreshold |= pendingProcessedBytes < thresholdBytes;
        }
        afterPendingUpdated();
    }

    @Override
    public void messageReceived(Message msg) {
        pendingReceivedMessages = Math.max(0, pendingReceivedMessages - 1);
        if (pendingReceivedMessages == 0) {
            noReceivedArePending = true;
        }
        if (isTrackingBytes) {
            pendingReceivedBytes = Math.max(0, pendingReceivedBytes - msg.consumeByteCount());
            noReceivedArePending |= pendingReceivedBytes == 0;
        }
        afterPendingUpdated();
    }

    @Override
    public void pullCompletedWithStatus(int messages, long bytes) {
        if (messages == -1) {
            // status without Nats-Pending-* headers
            fullResetPending();
        }
        else {
            statusAdjustPending(messages, bytes);
        }
        afterPendingUpdated();
    }

    @Override
    public void pullTerminatedByError() {
        if (stopped.get()) {
            fullClose();
        }
        else {
            try {
                shutdownSub();
                doSub(false);
            }
            catch (JetStreamException e) {
                resetOnException();
            }
            catch (InterruptedException e) {
                // Runs on a NATS-owned thread (heartbeat timer pool or message delivery) that
                // can't propagate the checked interrupt. Realistically only our own close
                // (shutdownNow) interrupts here, and the stopped/fullClose path handles that
                // teardown; a stray interrupt is transient, so recovering is correct. Restore
                // the flag as standard hygiene (a no-op on a pooled worker), then recover.
                Thread.currentThread().interrupt();
                resetOnException();
            }
        }
    }

    void doSub(boolean first) throws JetStreamException, InterruptedException {
        MessageHandler mh = userMessageHandler == null ? null : msg -> {
            try {
                userMessageHandler.onMessage(msg);
            }
            finally {
                updateProcessed(msg);
            }
        };

        // Whether the subscribe below succeeded, which is what decides if this attempt owns a
        // subscription that has to be cleaned up when something after it fails. It cannot be inferred
        // from the "sub" field: on the reset path that still refers to the previous, already shut down
        // subscription, and shutdownSub() dereferences it without a null check.
        boolean subInitialized = false;
        try {
            stopped.set(false);
            finished.set(false);
            super.initSub(subscriptionMaker.subscribe(mh, userDispatcher, pmm, null), !first);
            subInitialized = true;
            fullResetPending();
            rePull();
        }
        catch (JetStreamException | RuntimeException e) {
            if (subInitialized) {
                // The "subscribe" succeeded and something after it did not, so this attempt is being
                // abandoned while holding a live subscription. Nothing else will clean it up - on the
                // first == true path the consumer never reaches the caller, so both the subscription
                // and its heartbeat timer would be unreachable and leak.
                shutdownSub();
            }
            // The unchecked case matters as much as the checked one. A connection or dispatcher that is
            // closing or draining throws IllegalStateException out of the subscribe, and on the reset
            // path (first == false) shutdownSub() has already run. Letting that escape leaves the
            // consumer with no subscription and no heartbeat timer while stopped and finished both
            // still report false - it silently stops delivering, permanently. Recovering is correct:
            // resetOnException re-arms the heartbeat, which alarms again and retries the reset.
            if (first) {
                throw e;
            }
            resetOnException();
        }
    }

    private void resetOnException() {
        fullResetPending();
        pmm.updateLastMessageReceived();
        pmm.initOrResetHeartbeatTimer();
    }

    protected void afterPendingUpdated() {
        if (stopped.get()) {
            pmm.shutdownHeartbeatTimer();
            if (noReceivedArePending) {
                fullClose();
            }
        }
        else if (forcePull || processedHasCrossedThreshold) {
            rePull();
        }
    }

    protected void rePull() {
        int rePullMessages = Math.max(1, consumeOpts.getBatchSize() - pendingProcessedMessages);
        long rePullBytes = consumeOpts.getBatchBytes() == 0 ? 0 : consumeOpts.getBatchBytes() - pendingProcessedBytes;
        PinnablePullRequestOptions pro = new PinnablePullRequestOptions(pmm.currentPinId,
            PullRequestOptions.builder(rePullMessages)
                .maxBytes(rePullBytes)
                .expiresIn(consumeOpts.getExpiresInMillis())
                .idleHeartbeat(consumeOpts.getIdleHeartbeat())
                .group(consumeOpts.getGroup())
                .priority(consumeOpts.getPriority())
                .minPending(consumeOpts.getMinPending())
                .minAckPending(consumeOpts.getMinAckPending()));
        aboutToPull(rePullMessages, rePullBytes);
        sub._pull(pro, consumeOpts.raiseStatusWarnings(), this);
    }
}

1	package io.synadia.client.impl;
2
3	import io.synadia.client.Message;
4	import io.synadia.client.MessageHandler;
5	import io.synadia.client.api.ConsumerInfo;
6	import io.synadia.client.api.JetStreamException;
7
8
9	class NatsMessageConsumer extends NatsMessageConsumerBase implements PullManagerObserver {
10	protected final ConsumeOptions consumeOpts;
11	protected final SimplifiedSubscriptionMaker subscriptionMaker;
12	protected final NatsDispatcher userDispatcher;
13	protected final MessageHandler userMessageHandler;
14
15	protected final int thresholdMessages;
16	protected final long thresholdBytes;
17	protected final boolean isTrackingBytes;
18
19	protected int pendingReceivedMessages;
20	protected long pendingReceivedBytes;
21	protected boolean noReceivedArePending;
22	protected boolean forcePull;
23	protected int pendingProcessedMessages;
24	protected long pendingProcessedBytes;
25	protected boolean processedHasCrossedThreshold;
26
27	NatsMessageConsumer(SimplifiedSubscriptionMaker subscriptionMaker,
28	ConsumerInfo cachedConsumerInfo,
29	ConsumeOptions consumeOpts,
30	NatsDispatcher userDispatcher,
31	final MessageHandler userMessageHandler) throws JetStreamException, InterruptedException
32	{
33	super(cachedConsumerInfo);	3 ✔
34
35	this.subscriptionMaker = subscriptionMaker;	3 ✔
36	this.consumeOpts = consumeOpts;	3 ✔
37	this.userDispatcher = userDispatcher;	3 ✔
38	this.userMessageHandler = userMessageHandler;	3 ✔
39
40	int bm = consumeOpts.getBatchSize();	3 ✔
41	long bb = consumeOpts.getBatchBytes();	3 ✔
42	int rePullMessages = Math.max(1, bm * consumeOpts.getThresholdPercent() / 100);	9 ✔
43	long rePullBytes = bb == 0 ? 0 : Math.max(1, bb * consumeOpts.getThresholdPercent() / 100);	7 !
44	// The threshold is what the pending count has to drop below for the next pull. A batch of 1
45	// gives a re-pull size of 1 and so a threshold of 0, which pending can never go below, so the
46	// consumer pulled once and then stopped. Never let the threshold fall under 1.
47	int thresholdM = bm - rePullMessages;	4 ✔
48	thresholdMessages = thresholdM < 1 ? 1 : thresholdM;	8 ✔
49	long thresholdB = bb - rePullBytes;	4 ✔
50	thresholdBytes = bb == 0 ? Integer.MIN_VALUE : (thresholdB < 1 ? 1 : thresholdB);	8 !
51	isTrackingBytes = rePullBytes > 0;	7 !
52	doSub(true);	3 ✔
53	}	1 ✔
54
55	protected void fullResetPending() {
56	pendingReceivedMessages = 0;	3 ✔
57	pendingReceivedBytes = 0;	3 ✔
58	noReceivedArePending = true;	3 ✔
59	forcePull = true;	3 ✔
60	pendingProcessedMessages = 0;	3 ✔
61	pendingProcessedBytes = 0;	3 ✔
62	processedHasCrossedThreshold = true;	3 ✔
63	}	1 ✔
64
65	protected void statusAdjustPending(int messages, long bytes) {
66	pendingReceivedMessages = Math.max(0, pendingReceivedMessages - messages);	8 ✔
67	pendingReceivedBytes = Math.max(0, pendingReceivedBytes - bytes);	8 ✔
68	noReceivedArePending = pendingReceivedMessages == 0 \|\| (isTrackingBytes && pendingReceivedBytes == 0);	11 !
69	forcePull = true;	3 ✔
70	pendingProcessedMessages = Math.max(0, pendingProcessedMessages - messages);	8 ✔
71	pendingProcessedBytes = Math.max(0, pendingProcessedBytes - bytes);	8 ✔
72	processedHasCrossedThreshold = pendingProcessedMessages < thresholdMessages \|\| (isTrackingBytes && pendingProcessedBytes < thresholdBytes);	13 !
73	}	1 ✔
74
75	protected void aboutToPull(int messages, long bytes) {
76	pendingReceivedMessages += messages;	6 ✔
77	pendingReceivedBytes += bytes;	6 ✔
78	noReceivedArePending = false;	3 ✔
79	forcePull = false;	3 ✔
80	pendingProcessedMessages += messages;	6 ✔
81	pendingProcessedBytes += bytes;	6 ✔
82	processedHasCrossedThreshold = false;	3 ✔
83	}	1 ✔
84
85	protected void updateProcessed(Message msg) {
86	pendingProcessedMessages = Math.max(0, pendingProcessedMessages - 1);	8 ✔
87	if (pendingProcessedMessages < thresholdMessages) {	5 ✔
88	processedHasCrossedThreshold = true;	3 ✔
89	}
90	if (isTrackingBytes) {	3 !
91	pendingProcessedBytes = Math.max(0, pendingProcessedBytes - msg.consumeByteCount());	×
92	processedHasCrossedThreshold \|= pendingProcessedBytes < thresholdBytes;	×
93	}
94	afterPendingUpdated();	2 ✔
95	}	1 ✔
96
97	@Override
98	public void messageReceived(Message msg) {
99	pendingReceivedMessages = Math.max(0, pendingReceivedMessages - 1);	8 ✔
100	if (pendingReceivedMessages == 0) {	3 ✔
101	noReceivedArePending = true;	3 ✔
102	}
103	if (isTrackingBytes) {	3 !
104	pendingReceivedBytes = Math.max(0, pendingReceivedBytes - msg.consumeByteCount());	×
105	noReceivedArePending \|= pendingReceivedBytes == 0;	×
106	}
107	afterPendingUpdated();	2 ✔
108	}	1 ✔
109
110	@Override
111	public void pullCompletedWithStatus(int messages, long bytes) {
112	if (messages == -1) {	3 !
113	// status without Nats-Pending-* headers
114	fullResetPending();	×
115	}
116	else {
117	statusAdjustPending(messages, bytes);	4 ✔
118	}
119	afterPendingUpdated();	2 ✔
120	}	1 ✔
121
122	@Override
123	public void pullTerminatedByError() {
124	if (stopped.get()) {	4 ✔
125	fullClose();	3 ✔
126	}
127	else {
128	try {
129	shutdownSub();	2 ✔
130	doSub(false);	3 ✔
131	}
132	catch (JetStreamException e) {	×
133	resetOnException();	×
134	}
135	catch (InterruptedException e) {	×
136	// Runs on a NATS-owned thread (heartbeat timer pool or message delivery) that
137	// can't propagate the checked interrupt. Realistically only our own close
138	// (shutdownNow) interrupts here, and the stopped/fullClose path handles that
139	// teardown; a stray interrupt is transient, so recovering is correct. Restore
140	// the flag as standard hygiene (a no-op on a pooled worker), then recover.
141	Thread.currentThread().interrupt();	×
142	resetOnException();	×
143	}	1 ✔
144	}
145	}	1 ✔
146
147	void doSub(boolean first) throws JetStreamException, InterruptedException {
148	MessageHandler mh = userMessageHandler == null ? null : msg -> {	8 ✔
149	try {
150	userMessageHandler.onMessage(msg);	4 ✔
151	}
152	finally {
153	updateProcessed(msg);	3 ✔
154	}
155	};	1 ✔
156
157	// Whether the subscribe below succeeded, which is what decides if this attempt owns a
158	// subscription that has to be cleaned up when something after it fails. It cannot be inferred
159	// from the "sub" field: on the reset path that still refers to the previous, already shut down
160	// subscription, and shutdownSub() dereferences it without a null check.
161	boolean subInitialized = false;	2 ✔
162	try {
163	stopped.set(false);	4 ✔
164	finished.set(false);	4 ✔
165	super.initSub(subscriptionMaker.subscribe(mh, userDispatcher, pmm, null), !first);	16 ✔
166	subInitialized = true;	2 ✔
167	fullResetPending();	2 ✔
168	rePull();	2 ✔
169	}
170	catch (JetStreamException \| RuntimeException e) {	1 ✔
171	if (subInitialized) {	2 ✔
172	// The "subscribe" succeeded and something after it did not, so this attempt is being
173	// abandoned while holding a live subscription. Nothing else will clean it up - on the
174	// first == true path the consumer never reaches the caller, so both the subscription
175	// and its heartbeat timer would be unreachable and leak.
176	shutdownSub();	2 ✔
177	}
178	// The unchecked case matters as much as the checked one. A connection or dispatcher that is
179	// closing or draining throws IllegalStateException out of the subscribe, and on the reset
180	// path (first == false) shutdownSub() has already run. Letting that escape leaves the
181	// consumer with no subscription and no heartbeat timer while stopped and finished both
182	// still report false - it silently stops delivering, permanently. Recovering is correct:
183	// resetOnException re-arms the heartbeat, which alarms again and retries the reset.
184	if (first) {	2 ✔
185	throw e;	2 ✔
186	}
187	resetOnException();	2 ✔
188	}	1 ✔
189	}	1 ✔
190
191	private void resetOnException() {
192	fullResetPending();	2 ✔
193	pmm.updateLastMessageReceived();	3 ✔
194	pmm.initOrResetHeartbeatTimer();	3 ✔
195	}	1 ✔
196
197	protected void afterPendingUpdated() {
198	if (stopped.get()) {	4 ✔
199	pmm.shutdownHeartbeatTimer();	3 ✔
200	if (noReceivedArePending) {	3 ✔
201	fullClose();	3 ✔
202	}
203	}
204	else if (forcePull \|\| processedHasCrossedThreshold) {	6 ✔
205	rePull();	2 ✔
206	}
207	}	1 ✔
208
209	protected void rePull() {
210	int rePullMessages = Math.max(1, consumeOpts.getBatchSize() - pendingProcessedMessages);	9 ✔
211	long rePullBytes = consumeOpts.getBatchBytes() == 0 ? 0 : consumeOpts.getBatchBytes() - pendingProcessedBytes;	9 !
212	PinnablePullRequestOptions pro = new PinnablePullRequestOptions(pmm.currentPinId,	6 ✔
213	PullRequestOptions.builder(rePullMessages)	2 ✔
214	.maxBytes(rePullBytes)	3 ✔
215	.expiresIn(consumeOpts.getExpiresInMillis())	4 ✔
216	.idleHeartbeat(consumeOpts.getIdleHeartbeat())	4 ✔
217	.group(consumeOpts.getGroup())	4 ✔
218	.priority(consumeOpts.getPriority())	4 ✔
219	.minPending(consumeOpts.getMinPending())	4 ✔
220	.minAckPending(consumeOpts.getMinAckPending()));	4 ✔
221	aboutToPull(rePullMessages, rePullBytes);	4 ✔
222	sub._pull(pro, consumeOpts.raiseStatusWarnings(), this);	9 ✔
223	}	1 ✔
224	}

synadia-io / nats.java.v3 / 36783609042

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