Skip to content

Stop tokenRenewalWorks from leaking clients when it fails #1861

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
Ewerton Scaboro da Silva (ewertons) merged 2 commits into from
Aug 19, 2026
+161 −54
Merged

Stop tokenRenewalWorks from leaking clients when it fails #1861

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
74dde8e
Stop tokenRenewalWorks from leaking clients when it fails
ewertons Aug 19, 2026
99e506f
Clean up when the test fails during setup as well
ewertons Aug 19, 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
215 changes: 161 additions & 54 deletions ...src/test/java/tests/integration/com/microsoft/azure/sdk/iot/iothub/TokenRenewalTests.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -113,70 +113,165 @@ public static void stopProxy()
//@ContinuousIntegrationTest
public void tokenRenewalWorks() throws Exception
{
List<InternalClient> clients = createClientsToTest();
String hostname = clients.get(0).getConfig().getIotHubHostname();
// These are declared before the try so that the cleanup below can reach whatever was created before a failure.
// Creating the clients registers identities as it goes, so a failure part way through creation has something
// to clean up just as much as a failure during the test body does.
List<InternalClient> clients = new ArrayList<>();
ArrayList<DeviceClient> amqpMultiplexedClients = new ArrayList<>();
MultiplexingClient amqpMultiplexingClient = createMultiplexedClientToTest(AMQPS, amqpMultiplexedClients, hostname);
ArrayList<DeviceClient> amqpwsMultiplexedClients = new ArrayList<>();
MultiplexingClient amqpWsMultiplexingClient = createMultiplexedClientToTest(AMQPS_WS, amqpwsMultiplexedClients, hostname);
MultiplexingClient amqpMultiplexingClient = null;
MultiplexingClient amqpWsMultiplexingClient = null;

// Allow registry operations some buffer time before attempting to open connections for them
Thread.sleep(2000);
try
Comment thread
ewertons marked this conversation as resolved.
{
createClientsToTest(clients);
String hostname = clients.get(0).getConfig().getIotHubHostname();
amqpMultiplexingClient = createMultiplexedClientToTest(AMQPS, amqpMultiplexedClients, hostname);
amqpWsMultiplexingClient = createMultiplexedClientToTest(AMQPS_WS, amqpwsMultiplexedClients, hostname);

// Allow registry operations some buffer time before attempting to open connections for them
Thread.sleep(2000);

//service grants a 10 minute grace period beyond when sas token expires, this test attempts to send a message after that grace period
// to ensure that the first sas token has expired, and that the sas token was renewed successfully.
final long WAIT_BUFFER_FOR_TOKEN_TO_EXPIRE = EXPIRED_SAS_TOKEN_GRACE_PERIOD_SECONDS + EXTRA_BUFFER_TO_ENSURE_TOKEN_EXPIRED_SECONDS;

clients.addAll(amqpMultiplexedClients);
clients.addAll(amqpwsMultiplexedClients);

Success[] amqpDisconnectDidNotHappenSuccesses = new Success[clients.size()];
Success[] mqttDisconnectDidHappenSuccesses = new Success[clients.size()];
Success[] shutdownWasGracefulSuccesses = new Success[clients.size()];
Success[] mqttDisconnectHadTokenExpiredReasonSuccesses = new Success[clients.size()];
for (int clientIndex = 0; clientIndex < clients.size(); clientIndex++)
{
amqpDisconnectDidNotHappenSuccesses[clientIndex] = new Success();
mqttDisconnectDidHappenSuccesses[clientIndex] = new Success();
shutdownWasGracefulSuccesses[clientIndex] = new Success();
mqttDisconnectHadTokenExpiredReasonSuccesses[clientIndex] = new Success();

amqpDisconnectDidNotHappenSuccesses[clientIndex].setResult(true); //assume success until unexpected DISCONNECTED_RETRYING
mqttDisconnectDidHappenSuccesses[clientIndex].setResult(false); //assume failure until DISCONNECTED_RETRYING is triggered by token expiring
shutdownWasGracefulSuccesses[clientIndex].setResult(true); //assume success until DISCONNECTED callback without CLIENT_CLOSE

mqttDisconnectHadTokenExpiredReasonSuccesses[clientIndex].setResult(false); //assume failure until first disconnected_retrying executes with reason EXPIRED_SAS_TOKEN

clients.get(clientIndex).setConnectionStatusChangeCallback(
new IotHubConnectionStatusChangeTokenRenewalCallbackVerifier(
clients.get(clientIndex).getConfig().getProtocol(),
amqpDisconnectDidNotHappenSuccesses[clientIndex],
mqttDisconnectDidHappenSuccesses[clientIndex],
shutdownWasGracefulSuccesses[clientIndex],
mqttDisconnectHadTokenExpiredReasonSuccesses[clientIndex]),
clients.get(clientIndex));
}

//service grants a 10 minute grace period beyond when sas token expires, this test attempts to send a message after that grace period
// to ensure that the first sas token has expired, and that the sas token was renewed successfully.
final long WAIT_BUFFER_FOR_TOKEN_TO_EXPIRE = EXPIRED_SAS_TOKEN_GRACE_PERIOD_SECONDS + EXTRA_BUFFER_TO_ENSURE_TOKEN_EXPIRED_SECONDS;
openEachClient(clients);
amqpMultiplexingClient.open(false);
amqpWsMultiplexingClient.open(false);

clients.addAll(amqpMultiplexedClients);
clients.addAll(amqpwsMultiplexedClients);
//wait until old sas token has expired, this should force the config to generate a new one from the device key
System.out.println("Sleeping..." + System.currentTimeMillis());
Thread.sleep((SECONDS_FOR_SAS_TOKEN_TO_LIVE_BEFORE_RENEWAL + WAIT_BUFFER_FOR_TOKEN_TO_EXPIRE) * 1000);
System.out.println("Awake!" + System.currentTimeMillis());

Success[] amqpDisconnectDidNotHappenSuccesses = new Success[clients.size()];
Success[] mqttDisconnectDidHappenSuccesses = new Success[clients.size()];
Success[] shutdownWasGracefulSuccesses = new Success[clients.size()];
Success[] mqttDisconnectHadTokenExpiredReasonSuccesses = new Success[clients.size()];
for (int clientIndex = 0; clientIndex < clients.size(); clientIndex++)
sendMessageFromEachClient(clients);

closeClients(clients);
amqpMultiplexingClient.close();
amqpWsMultiplexingClient.close();

Tools.disposeTestIdentities(testIdentities, iotHubConnectionString);

testIdentities.clear();

verifyClientsConnectivityBehavedCorrectly(clients, amqpDisconnectDidNotHappenSuccesses, mqttDisconnectDidHappenSuccesses, shutdownWasGracefulSuccesses, mqttDisconnectHadTokenExpiredReasonSuccesses);
}
finally
{
amqpDisconnectDidNotHappenSuccesses[clientIndex] = new Success();
mqttDisconnectDidHappenSuccesses[clientIndex] = new Success();
shutdownWasGracefulSuccesses[clientIndex] = new Success();
mqttDisconnectHadTokenExpiredReasonSuccesses[clientIndex] = new Success();

amqpDisconnectDidNotHappenSuccesses[clientIndex].setResult(true); //assume success until unexpected DISCONNECTED_RETRYING
mqttDisconnectDidHappenSuccesses[clientIndex].setResult(false); //assume failure until DISCONNECTED_RETRYING is triggered by token expiring
shutdownWasGracefulSuccesses[clientIndex].setResult(true); //assume success until DISCONNECTED callback without CLIENT_CLOSE

mqttDisconnectHadTokenExpiredReasonSuccesses[clientIndex].setResult(false); //assume failure until first disconnected_retrying executes with reason EXPIRED_SAS_TOKEN

clients.get(clientIndex).setConnectionStatusChangeCallback(
new IotHubConnectionStatusChangeTokenRenewalCallbackVerifier(
clients.get(clientIndex).getConfig().getProtocol(),
amqpDisconnectDidNotHappenSuccesses[clientIndex],
mqttDisconnectDidHappenSuccesses[clientIndex],
shutdownWasGracefulSuccesses[clientIndex],
mqttDisconnectHadTokenExpiredReasonSuccesses[clientIndex]),
clients.get(clientIndex));
// Without this, a failure part way through the test leaves behind every client that had already been
// opened. Those clients keep retrying their connections for the rest of the JVM's life, and the ones that
// were given proxy settings keep retrying through the proxy this class runs locally. That competes with
// the reruns of this test for the agent's resources and for that proxy, which makes the reruns fail the
// same way the first attempt did rather than giving them a clean chance to pass.
closeClientsQuietly(clients);

// The multiplexed clients are cleaned up separately because they are only added to the list above once
// both multiplexing clients have been created. A failure before that point leaves them reachable only
// through these lists.
closeClientsQuietly(amqpMultiplexedClients);
closeClientsQuietly(amqpwsMultiplexedClients);

closeClientQuietly(amqpMultiplexingClient);
closeClientQuietly(amqpWsMultiplexingClient);

// Identities are removed here as well so that a failed run does not leave them behind in the registry.
// This is a no-op when the test succeeded, because the successful path already disposed of them.
disposeTestIdentitiesQuietly();
}
}

openEachClient(clients);
amqpMultiplexingClient.open(false);
amqpWsMultiplexingClient.open(false);
/**
* Close each client, ignoring any failure. This is only used while cleaning up, where a client that cannot be
* closed must not replace the failure that the test is already reporting.
*
* @param clients The clients to close
*/
private void closeClientsQuietly(List<? extends InternalClient> clients)
{
for (InternalClient client : clients)
{
closeClientQuietly(client);
}
}

//wait until old sas token has expired, this should force the config to generate a new one from the device key
System.out.println("Sleeping..." + System.currentTimeMillis());
Thread.sleep((SECONDS_FOR_SAS_TOKEN_TO_LIVE_BEFORE_RENEWAL + WAIT_BUFFER_FOR_TOKEN_TO_EXPIRE) * 1000);
System.out.println("Awake!" + System.currentTimeMillis());
private void closeClientQuietly(InternalClient client)
{
try
{
client.close();
}
catch (UnsupportedOperationException ex)
{
// Multiplexed clients throw this when closed through the individual client itself. They are closed by
// closing the multiplexing client that they belong to instead.
}
catch (Exception ex)
{
log.debug("Failed to close a client while cleaning up after the test", ex);
}
}

sendMessageFromEachClient(clients);
private void closeClientQuietly(MultiplexingClient multiplexingClient)
{
if (multiplexingClient == null)
{
// Creation did not get as far as this client
return;
}

closeClients(clients);
amqpMultiplexingClient.close();
amqpWsMultiplexingClient.close();
try
{
multiplexingClient.close();
}
catch (Exception ex)
{
log.debug("Failed to close a multiplexing client while cleaning up after the test", ex);
}
}

Tools.disposeTestIdentities(testIdentities, iotHubConnectionString);
private void disposeTestIdentitiesQuietly()
{
try
{
Tools.disposeTestIdentities(testIdentities, iotHubConnectionString);
}
catch (Exception ex)
{
log.debug("Failed to dispose of the test identities while cleaning up after the test", ex);
}

testIdentities.clear();

verifyClientsConnectivityBehavedCorrectly(clients, amqpDisconnectDidNotHappenSuccesses, mqttDisconnectDidHappenSuccesses, shutdownWasGracefulSuccesses, mqttDisconnectHadTokenExpiredReasonSuccesses);
}

private void closeClients(List<InternalClient> clients) throws IOException
Expand Down Expand Up @@ -221,9 +316,17 @@ else if (protocol == AMQPS || protocol == AMQPS_WS)
}
}

private List<InternalClient> createClientsToTest() throws IotHubException, IOException, URISyntaxException, InterruptedException, GeneralSecurityException
/**
* Create the clients that this test exercises, adding each one to the given list as it is created.
*
* <p>The clients are added to the caller's list rather than returned all at once so that a failure part way
* through creation still leaves the caller holding the clients that were created before it, which lets them be
* cleaned up rather than leaked.</p>
*
* @param clients The list to add the created clients to
*/
private void createClientsToTest(List<InternalClient> clients) throws IotHubException, IOException, URISyntaxException, InterruptedException, GeneralSecurityException
{
List<InternalClient> clients = new ArrayList<>();
Proxy testProxy = new Proxy(Proxy.Type.HTTP, new InetSocketAddress(testProxyHostname, testProxyPort));
for (IotHubClientProtocol protocol: IotHubClientProtocol.values())
{
Expand All @@ -248,10 +351,14 @@ private List<InternalClient> createClientsToTest() throws IotHubException, IOExc
Device device = new Device(deviceId);
device = registryClient.addDevice(device);
SasTokenProvider sasTokenProvider = new SasTokenProviderImpl(Tools.getDeviceConnectionString(iotHubConnectionString, device), SECONDS_FOR_SAS_TOKEN_TO_LIVE_BEFORE_RENEWAL);
clients.add(new DeviceClient(iotHubHostName, deviceId, sasTokenProvider, protocol));
}
DeviceClient clientWithCustomSasTokenProvider = new DeviceClient(iotHubHostName, deviceId, sasTokenProvider, protocol);

return clients;
// This device is registered here rather than handed out by Tools, so it has to be tracked alongside the
// other identities. Without this it is never disposed of and is left behind in the registry on every run,
// whether the test passes or fails.
testIdentities.add(new TestDeviceIdentity(clientWithCustomSasTokenProvider, device));
clients.add(clientWithCustomSasTokenProvider);
}
}

private MultiplexingClient createMultiplexedClientToTest(IotHubClientProtocol protocol, List<DeviceClient> clientsToCreate, String hostname) throws IotHubException, IOException, URISyntaxException, InterruptedException, GeneralSecurityException, IotHubClientException, TimeoutException
Expand Down