#2035

Committed 03 Jul 2025 12:14PM UTC coverage: 95.611% (-0.04%) from 95.651%

Build # #2035

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #1345 from nats-io/debug-unit-test-hanging-01

[Bug] Pull Heartbeat handler intermittent failure after switch to scheduler

Run Details

57 of 68 new or added lines in 5 files covered. (83.82%)

3 existing lines in 2 files now uncovered.

11786 of 12327 relevant lines covered (95.61%)

0.96 hits per line

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

69.57

/src/main/java/io/nats/client/support/ScheduledTask.java

// Copyright 2025 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.support;

import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;


/**
 * !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! *
 * WARNING: THIS CLASS IS PUBLIC BUT ITS API IS NOT GUARANTEED TO *
 * BE BACKWARD COMPATIBLE AS IT IS INTENDED AS AN INTERNAL CLASS  *
 * !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! *
 */
public class ScheduledTask implements Runnable {
    private static final AtomicLong ID_GENERATOR = new AtomicLong();

    private final String id;
    private final Runnable runnable;
    protected final AtomicReference<ScheduledFuture<?>> scheduledFutureRef;

    protected final AtomicBoolean notShutdown;
    protected final AtomicBoolean executing;

    public ScheduledTask(ScheduledExecutorService ses, long initialAndPeriodMillis, Runnable runnable) {
        this(null, ses, initialAndPeriodMillis, initialAndPeriodMillis, TimeUnit.MILLISECONDS, runnable);
    }

    public ScheduledTask(String id, ScheduledExecutorService ses, long initialAndPeriodMillis, Runnable runnable) {
        this(id, ses, initialAndPeriodMillis, initialAndPeriodMillis, TimeUnit.MILLISECONDS, runnable);
    }

    public ScheduledTask(ScheduledExecutorService ses, long initialAndPeriod, TimeUnit unit, Runnable runnable) {
        this(null, ses, initialAndPeriod, initialAndPeriod, unit, runnable);
    }

    public ScheduledTask(String id, ScheduledExecutorService ses, long initialAndPeriod, TimeUnit unit, Runnable runnable) {
        this(id, ses, initialAndPeriod, initialAndPeriod, unit, runnable);
    }

    public ScheduledTask(ScheduledExecutorService ses, long initialDelay, long initialAndPeriod, TimeUnit unit, Runnable runnable) {
        this(null, ses, initialDelay, initialAndPeriod, unit, runnable);
    }

    public ScheduledTask(String id, ScheduledExecutorService ses, long initialDelay, long initialAndPeriod, TimeUnit unit, Runnable runnable) {
        this.id = id == null || id.isEmpty() ? "st-" + ID_GENERATOR.getAndIncrement() : id;
        this.runnable = runnable;
        notShutdown = new AtomicBoolean(true);
        executing = new AtomicBoolean(false);
        scheduledFutureRef = new AtomicReference<>(
            ses.scheduleAtFixedRate(this, initialDelay, initialAndPeriod, unit));
    }

    @Override
    public void run() {
        try {
            if (notShutdown.get()) {
                executing.set(true);
                runnable.run();
            }
        }
        finally {
            executing.set(false);
        }
    }

    public boolean isShutdown() {
        return !notShutdown.get();
    }

    public boolean isExecuting() {
        return executing.get();
    }

    public boolean isDone() {
        ScheduledFuture<?> f = scheduledFutureRef.get();
        return f == null || f.isDone();
    }

    public String getId() {
        return id;
    }

    public void shutdown() {
        try {
            notShutdown.set(false);
            ScheduledFuture<?> f = scheduledFutureRef.get();
            if (f != null) {
                scheduledFutureRef.set(null); // just releasing resources.
                if (!f.isDone()) {
                    f.cancel(false);
                }
            }
        }
        catch (Exception ignore) {
            // don't want this to be passed along
        }
    }

    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder(id);
        if (notShutdown.get()) {
            sb.append(" [live");
        }
        else {
            sb.append(" [shutdown");
        }
        sb.append(isDone() ? "/done" : "/not done");
        sb.append(executing.get() ? "/executing" : "/not executing");
        sb.append("]");
        return sb.toString();
    }
}

1	// Copyright 2025 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.support;
15
16	import java.util.concurrent.ScheduledExecutorService;
17	import java.util.concurrent.ScheduledFuture;
18	import java.util.concurrent.TimeUnit;
19	import java.util.concurrent.atomic.AtomicBoolean;
20	import java.util.concurrent.atomic.AtomicLong;
21	import java.util.concurrent.atomic.AtomicReference;
22
23
24	/**
25	* !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! *
26	* WARNING: THIS CLASS IS PUBLIC BUT ITS API IS NOT GUARANTEED TO *
27	* BE BACKWARD COMPATIBLE AS IT IS INTENDED AS AN INTERNAL CLASS *
28	* !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! *
29	*/
30	public class ScheduledTask implements Runnable {
31	private static final AtomicLong ID_GENERATOR = new AtomicLong();	1✔
32
33	private final String id;
34	private final Runnable runnable;
35	protected final AtomicReference<ScheduledFuture<?>> scheduledFutureRef;
36
37	protected final AtomicBoolean notShutdown;
38	protected final AtomicBoolean executing;
39
40	public ScheduledTask(ScheduledExecutorService ses, long initialAndPeriodMillis, Runnable runnable) {
41	this(null, ses, initialAndPeriodMillis, initialAndPeriodMillis, TimeUnit.MILLISECONDS, runnable);	1✔
42	}	1✔
43
44	public ScheduledTask(String id, ScheduledExecutorService ses, long initialAndPeriodMillis, Runnable runnable) {
NEW 45	this(id, ses, initialAndPeriodMillis, initialAndPeriodMillis, TimeUnit.MILLISECONDS, runnable);	×
UNCOV 46	}	×
47
48	public ScheduledTask(ScheduledExecutorService ses, long initialAndPeriod, TimeUnit unit, Runnable runnable) {
49	this(null, ses, initialAndPeriod, initialAndPeriod, unit, runnable);	1✔
50	}	1✔
51
52	public ScheduledTask(String id, ScheduledExecutorService ses, long initialAndPeriod, TimeUnit unit, Runnable runnable) {
NEW 53	this(id, ses, initialAndPeriod, initialAndPeriod, unit, runnable);	×
UNCOV 54	}	×
55
56	public ScheduledTask(ScheduledExecutorService ses, long initialDelay, long initialAndPeriod, TimeUnit unit, Runnable runnable) {
57	this(null, ses, initialDelay, initialAndPeriod, unit, runnable);	1✔
58	}	1✔
59
60	public ScheduledTask(String id, ScheduledExecutorService ses, long initialDelay, long initialAndPeriod, TimeUnit unit, Runnable runnable) {	1✔
61	this.id = id == null \|\| id.isEmpty() ? "st-" + ID_GENERATOR.getAndIncrement() : id;	1✔
62	this.runnable = runnable;	1✔
63	notShutdown = new AtomicBoolean(true);	1✔
64	executing = new AtomicBoolean(false);	1✔
65	scheduledFutureRef = new AtomicReference<>(	1✔
66	ses.scheduleAtFixedRate(this, initialDelay, initialAndPeriod, unit));	1✔
67	}	1✔
68
69	@Override
70	public void run() {
71	try {
72	if (notShutdown.get()) {	1✔
73	executing.set(true);	1✔
74	runnable.run();	1✔
75	}
76	}
77	finally {
78	executing.set(false);	1✔
79	}
80	}	1✔
81
82	public boolean isShutdown() {
83	return !notShutdown.get();	1✔
84	}
85
86	public boolean isExecuting() {
87	return executing.get();	1✔
88	}
89
90	public boolean isDone() {
91	ScheduledFuture<?> f = scheduledFutureRef.get();	1✔
92	return f == null \|\| f.isDone();	1✔
93	}
94
95	public String getId() {
NEW 96	return id;	×
97	}
98
99	public void shutdown() {
100	try {
101	notShutdown.set(false);	1✔
102	ScheduledFuture<?> f = scheduledFutureRef.get();	1✔
103	if (f != null) {	1✔
104	scheduledFutureRef.set(null); // just releasing resources.	1✔
105	if (!f.isDone()) {	1✔
106	f.cancel(false);	1✔
107	}
108	}
109	}
110	catch (Exception ignore) {	×
111	// don't want this to be passed along
112	}	1✔
113	}	1✔
114
115	@Override
116	public String toString() {
NEW 117	StringBuilder sb = new StringBuilder(id);	×
NEW 118	if (notShutdown.get()) {	×
NEW 119	sb.append(" [live");	×
120	}
121	else {
NEW 122	sb.append(" [shutdown");	×
123	}
NEW 124	sb.append(isDone() ? "/done" : "/not done");	×
NEW 125	sb.append(executing.get() ? "/executing" : "/not executing");	×
NEW 126	sb.append("]");	×
NEW 127	return sb.toString();	×
128	}
129	}

nats-io / nats.java / #2035

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