Skip to content

[mqtt.ruuvigateway] Fix flaky itest and NPE on thing disposal (test teardown) #21360

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

Open
ssalonen wants to merge 2 commits into
base: main
Choose a base branch
from
Open

[mqtt.ruuvigateway] Fix flaky itest and NPE on thing disposal (test teardown) #21360

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
95c5fe6
[mqtt.ruuvigateway] Make itest robust against MQTT broker bridge state
claude Aug 11, 2026
6ef16c0
[mqtt.ruuvigateway] Fix NPE when unsubscribing on disposal
claude Aug 11, 2026
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
3 changes: 2 additions & 1 deletion ...ay/src/main/java/org/openhab/binding/mqtt/ruuvigateway/internal/handler/RuuviHandler.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -232,7 +232,8 @@ public CompletableFuture<Void> unsubscribeAll() {
MqttBrokerConnection localConnection = connection;
String localTopic = topic;
if (localConnection != null && localTopic != null) {
return localConnection.unsubscribe(localTopic, this).thenCompose(unsubscribeSuccessful -> null);
return localConnection.unsubscribe(localTopic, this).thenAccept(unsubscribeSuccessful -> {

@lsiepel lsiepel Aug 15, 2026

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.

Could we add a focused regression test for this completion-stage fix? The updated integration-test teardown only waits until thing.getHandler() is null. AbstractMQTTThingHandler.dispose() catches the ExecutionException produced by the old thenCompose(... -> null) implementation, after which the handler is still detached, so the old bug would satisfy the new assertion. A test should verify that unsubscribeAll() completes normally (or that the subscription is actually removed).

});
} else {
return CompletableFuture.completedFuture(null);
}
Expand Down
63 changes: 47 additions & 16 deletions ...vigateway.tests/src/main/java/org/openhab/binding/mqtt/ruuvigateway/RuuviGatewayTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,7 @@
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;

import javax.measure.quantity.Acceleration;
Expand Down Expand Up @@ -170,9 +171,14 @@ private Bridge createMqttBrokerBridge() {
Bridge bridge = BridgeBuilder.create(new ThingTypeUID("mqtt", "broker"), "mybroker").withLabel("MQTT Broker")
.withConfiguration(configuration).build();
thingProvider.add(bridge);
things.add(bridge);
waitForAssert(() -> assertNotNull(bridge.getHandler()));
assertNotNull(bridge.getConfiguration());
things.add(bridge);
// Wait until the broker bridge is really connected before any Ruuvi things are created. A Ruuvi thing that
// is initialized while the bridge is not ONLINE immediately reports OFFLINE (BRIDGE_OFFLINE) or
// OFFLINE (COMMUNICATION_ERROR), which would show up as an extra, unexpected thing status update in the
// assertions of the tests.
waitForAssert(() -> assertEquals(ThingStatus.ONLINE, bridge.getStatus(), bridge.getStatusInfo().toString()));
return bridge;
}

Expand Down Expand Up @@ -201,9 +207,9 @@ private Thing createRuuviThing(String brokerPrefix, String topic, @Nullable Inte

Thing thing = thingBuilder.build();
thingProvider.add(thing);
things.add(thing);
waitForAssert(() -> assertNotNull(thing.getHandler()));
assertNotNull(thing.getConfiguration());
things.add(thing);
return thing;
}

Expand Down Expand Up @@ -284,19 +290,35 @@ public void beforeEach() throws Exception {
@Override
@AfterEach
public void afterEach() throws Exception {
unregisterService(statusSubscriber);
removeThings();

@lsiepel lsiepel Aug 15, 2026

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.

If removeThings() times out or a removal throws, afterEach() exits here and never stops mqttConnection, shuts down the scheduler, or calls super.afterEach(). That leaves the very broker resources this change is intended to contain leaking into subsequent tests (and can mask the original failure). Please preserve the required thing-before-broker removal order, but protect the remaining cleanup with try/finally so it always runs.

if (mqttConnection != null) {
mqttConnection.removeConnectionObserver(failIfChange);
mqttConnection.stop().get(5, TimeUnit.SECONDS);
}
things.stream().map(thing -> thingProvider.remove(thing.getUID()));
unregisterService(statusSubscriber);

if (scheduler != null) {
scheduler.shutdownNow();
}
super.afterEach();
}

/**
* Remove the things created by the test and wait until their handlers have been disposed.
*
* This is done while the MQTT broker is still running. Without the explicit removal the things are removed only
* later on, when {@link org.openhab.core.test.java.JavaOSGiTest} unregisters the volatile storage service, i.e.
* after the broker has already been stopped. The broker connection of such a late disposed bridge handler keeps
* on reconnecting in the background and leaks into the next test.
*/
private void removeThings() {
// Remove the ruuvi things first, so that their handlers are disposed before the bridge handler
things.stream().filter(thing -> !(thing instanceof Bridge)).map(Thing::getUID).forEach(thingProvider::remove);
things.stream().filter(Bridge.class::isInstance).map(Thing::getUID).forEach(thingProvider::remove);
waitForAssert(() -> things.forEach(thing -> assertNull(thing.getHandler(), thing.getUID().getAsString())));
things.clear();
}

@Test
public void retrieveAllRuuviPrefixedTopics() throws Exception {
CountDownLatch c = new CountDownLatch(registeredTopics);
Expand All @@ -305,25 +327,34 @@ public void retrieveAllRuuviPrefixedTopics() throws Exception {
"Connection " + mqttConnection.getClientId() + " not retrieving all topics ");
}

/**
* Describe all status updates received so far. Since the tests assert the status updates by their index, the
* whole sequence is needed to make sense of a failed assertion. For example an unexpected
* OFFLINE (BRIDGE_OFFLINE) update in the sequence tells that the MQTT broker connection of the bridge was
* disturbed during the test.
*/
private Supplier<@Nullable String> describeStatusUpdates(List<ThingStatusInfo> statusUpdates, int index) {
return () -> String.format("Unexpected thing status update at index %d. Status updates received: %s", index,
statusUpdates.stream().map(ThingStatusInfo::toString).collect(Collectors.joining(", ", "[", "]")));
}

private void assertThingStatus(List<ThingStatusInfo> statusUpdates, int index, ThingStatus status,
@Nullable ThingStatusDetail detail, @Nullable String description) {
assertTrue(statusUpdates.size() > index,
String.format("Not enough status updates. Expected %d, but only had %d. Status updates received: %s",
index + 1, statusUpdates.size(),
statusUpdates.stream().map(ThingStatusInfo::getStatus).collect(Collectors.toList())));
assertEquals(status, statusUpdates.get(index).getStatus(), statusUpdates.get(index).toString());
assertEquals(detail, statusUpdates.get(index).getStatusDetail(), statusUpdates.get(index).toString());
assertEquals(description, statusUpdates.get(index).getDescription(), statusUpdates.get(index).toString());
Supplier<@Nullable String> message = describeStatusUpdates(statusUpdates, index);
assertTrue(statusUpdates.size() > index, message);
assertEquals(status, statusUpdates.get(index).getStatus(), message);
assertEquals(detail, statusUpdates.get(index).getStatusDetail(), message);
assertEquals(description, statusUpdates.get(index).getDescription(), message);
}

@SuppressWarnings("null")
private void assertThingStatusWithDescriptionPattern(List<ThingStatusInfo> statusUpdates, int index,
ThingStatus status, ThingStatusDetail detail, String descriptionPattern) {
assertTrue(statusUpdates.size() > index, "assert " + statusUpdates.size() + " > " + index + " failed");
assertEquals(status, statusUpdates.get(index).getStatus(), statusUpdates.get(index).toString());
assertEquals(detail, statusUpdates.get(index).getStatusDetail(), statusUpdates.get(index).toString());
assertTrue(statusUpdates.get(index).getDescription().matches(descriptionPattern),
statusUpdates.get(index).toString());
Supplier<@Nullable String> message = describeStatusUpdates(statusUpdates, index);
assertTrue(statusUpdates.size() > index, message);
assertEquals(status, statusUpdates.get(index).getStatus(), message);
assertEquals(detail, statusUpdates.get(index).getStatusDetail(), message);
assertTrue(statusUpdates.get(index).getDescription().matches(descriptionPattern), message);
}

private void assertThingStatus(List<ThingStatusInfo> statusUpdates, int index, ThingStatus status) {
Expand Down
Loading