Add continuation token support for SyncPoller ARM LROs

- Add PollingState.toContinuationToken() and fromContinuationToken() methods
- Add SyncPoller.serializeContinuationToken() interface method
- Implement ArmLroSyncPoller wrapper with continuation token support
- Add SyncPollerFactory.resumeFromToken() to reconstruct pollers from tokens
- Add comprehensive unit and integration tests (12 tests, all passing)
- Update CHANGELOGs for azure-core and azure-core-management

This enables ARM Long-Running Operations to be resumed across process
restarts by serializing the poller state to a Base64-encoded token.

Author: markli-db

sdk/core/azure-core-management/CHANGELOG.md

@@ -4,6 +4,13 @@
 
 ### Features Added
 
+- Added continuation token support for ARM Long-Running Operations (LROs):
+  - `PollingState.toContinuationToken()` - Serialize polling state to a Base64-encoded token
+  - `PollingState.fromContinuationToken()` - Deserialize token back to polling state
+  - `SyncPollerFactory.resumeFromToken()` - Resume a SyncPoller from a continuation token
+  - `ArmLroSyncPoller` - New implementation that supports continuation tokens for ARM LROs
+  - `PollerResumeContext` - Helper class to simplify resuming pollers by bundling infrastructure parameters
+
 ### Breaking Changes
 
 ### Bugs Fixed

sdk/core/azure-core-management/src/main/java/com/azure/core/management/implementation/polling/PollingState.java

@@ -169,6 +169,81 @@ public <T> void store(PollingContext<T> context) {
         }
     }
 
+    /**
+     * Serializes the current PollingState into a continuation token string that can be used to resume
+     * the long-running operation at a later time or in a different process.
+     * <p>
+     * The continuation token is a Base64-encoded JSON representation of the polling state. It contains
+     * all necessary information to reconstruct the poller and continue polling the operation, including:
+     * <ul>
+     * <li>Operation URL and HTTP method</li>
+     * <li>Current provisioning state</li>
+     * <li>Polling URLs (Azure-AsyncOperation, Location headers, etc.)</li>
+     * <li>Last known response data</li>
+     * </ul>
+     * <p>
+     * <strong>Security Note:</strong> The continuation token contains the operation URL. Ensure tokens
+     * are stored securely and transmitted over secure channels.
+     * <p>
+     * <strong>Compatibility Note:</strong> The token format is internal and may change between SDK versions.
+     * Tokens should only be used with the same version of the SDK that generated them.
+     *
+     * @return A Base64-encoded continuation token string representing the current polling state.
+     * @throws RuntimeException if the state cannot be serialized into a token.
+     */
+    public String toContinuationToken() {
+        try {
+            String jsonState = this.serializerAdapter.serialize(this, SerializerEncoding.JSON);
+            return java.util.Base64.getEncoder()
+                .encodeToString(jsonState.getBytes(java.nio.charset.StandardCharsets.UTF_8));
+        } catch (IOException ioe) {
+            throw LOGGER.logExceptionAsError(
+                new RuntimeException("Failed to serialize PollingState to continuation token.", ioe));
+        }
+    }
+
+    /**
+     * Deserializes a continuation token string into a PollingState object that can be used to resume
+     * a long-running operation.
+     * <p>
+     * This method is the counterpart to {@link #toContinuationToken()} and reconstructs a PollingState
+     * from a previously serialized token. The reconstructed state can then be used to create a new
+     * SyncPoller that continues polling from where the previous poller left off.
+     *
+     * @param continuationToken The Base64-encoded continuation token string, previously obtained from
+     *                          {@link #toContinuationToken()}.
+     * @param serializerAdapter The serializer for decoding the token. This should be the same serializer
+     *                          type used by the service client.
+     * @return A PollingState object reconstructed from the continuation token.
+     * @throws IllegalArgumentException if the {@code continuationToken} or {@code serializerAdapter} is null or empty.
+     * @throws RuntimeException if the token cannot be decoded or deserialized. This may occur if:
+     *         <ul>
+     *         <li>The token is malformed or corrupted</li>
+     *         <li>The token was created with a different SDK version</li>
+     *         <li>The token format has changed</li>
+     *         </ul>
+     */
+    public static PollingState fromContinuationToken(String continuationToken, SerializerAdapter serializerAdapter) {
+        Objects.requireNonNull(continuationToken, "'continuationToken' cannot be null.");
+        Objects.requireNonNull(serializerAdapter, "'serializerAdapter' cannot be null.");
+
+        if (continuationToken.isEmpty()) {
+            throw LOGGER.logExceptionAsError(new IllegalArgumentException("'continuationToken' cannot be empty."));
+        }
+
+        try {
+            byte[] decodedBytes = java.util.Base64.getDecoder().decode(continuationToken);
+            String jsonState = new String(decodedBytes, java.nio.charset.StandardCharsets.UTF_8);
+            return PollingState.from(serializerAdapter, jsonState);
+        } catch (IllegalArgumentException iae) {
+            throw LOGGER.logExceptionAsError(new RuntimeException(
+                "Failed to decode continuation token. The token may be malformed or corrupted.", iae));
+        } catch (RuntimeException re) {
+            throw LOGGER.logExceptionAsError(new RuntimeException("Failed to deserialize continuation token. "
+                + "The token may have been created with a different SDK version.", re));
+        }
+    }
+
     /**
      * @return the current status of the long-running-operation.
      */
@@ -277,7 +352,7 @@ FinalResult getFinalResult() {
     /**
      * @return the last response body this PollingState received
      */
-    String getLastResponseBody() {
+    public String getLastResponseBody() {
         return this.lastResponseBody;
     }
 

sdk/core/azure-core-management/src/main/java/com/azure/core/management/polling/ArmLroSyncPoller.java

@@ -0,0 +1,119 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.core.management.polling;
+
+import com.azure.core.management.implementation.polling.PollingState;
+import com.azure.core.util.logging.ClientLogger;
+import com.azure.core.util.polling.PollResponse;
+import com.azure.core.util.polling.PollingContext;
+import com.azure.core.util.polling.SyncPoller;
+import com.azure.core.util.serializer.SerializerAdapter;
+
+import java.time.Duration;
+import java.util.Objects;
+import java.util.function.BiFunction;
+import java.util.function.Function;
+
+/**
+ * Azure Resource Manager (ARM) Long-Running Operation SyncPoller implementation with continuation token support.
+ * <p>
+ * This implementation wraps a standard {@link SyncPoller} and adds ARM-specific functionality including
+ * the ability to serialize the poller state to a continuation token and resume from such a token.
+ * <p>
+ * This class is package-private and should only be created through {@link SyncPollerFactory}.
+ *
+ * @param <T> The type of poll response value.
+ * @param <U> The type of the final result of long-running operation.
+ */
+final class ArmLroSyncPoller<T, U> implements SyncPoller<PollResult<T>, U> {
+    private static final ClientLogger LOGGER = new ClientLogger(ArmLroSyncPoller.class);
+
+    private final SyncPoller<PollResult<T>, U> innerPoller;
+    private final SerializerAdapter serializerAdapter;
+    // We'll need a way to access the PollingContext from the inner poller
+    // Since SimpleSyncPoller doesn't expose it, we'll need to track it ourselves
+    private final PollingContextAccessor<PollResult<T>> contextAccessor;
+
+    /**
+     * Functional interface to access the PollingContext from a poller.
+     */
+    @FunctionalInterface
+    interface PollingContextAccessor<T> {
+        PollingContext<T> getContext();
+    }
+
+    /**
+     * Creates an ArmLroSyncPoller.
+     *
+     * @param innerPoller The underlying SyncPoller implementation.
+     * @param serializerAdapter The serializer for encoding/decoding.
+     * @param contextAccessor Accessor to get the current PollingContext.
+     */
+    ArmLroSyncPoller(SyncPoller<PollResult<T>, U> innerPoller, SerializerAdapter serializerAdapter,
+        PollingContextAccessor<PollResult<T>> contextAccessor) {
+        this.innerPoller = Objects.requireNonNull(innerPoller, "'innerPoller' cannot be null.");
+        this.serializerAdapter = Objects.requireNonNull(serializerAdapter, "'serializerAdapter' cannot be null.");
+        this.contextAccessor = Objects.requireNonNull(contextAccessor, "'contextAccessor' cannot be null.");
+    }
+
+    @Override
+    public PollResponse<PollResult<T>> poll() {
+        return innerPoller.poll();
+    }
+
+    @Override
+    public PollResponse<PollResult<T>> waitForCompletion() {
+        return innerPoller.waitForCompletion();
+    }
+
+    @Override
+    public PollResponse<PollResult<T>> waitForCompletion(Duration timeout) {
+        return innerPoller.waitForCompletion(timeout);
+    }
+
+    @Override
+    public PollResponse<PollResult<T>>
+        waitUntil(com.azure.core.util.polling.LongRunningOperationStatus statusToWaitFor) {
+        return innerPoller.waitUntil(statusToWaitFor);
+    }
+
+    @Override
+    public PollResponse<PollResult<T>> waitUntil(Duration timeout,
+        com.azure.core.util.polling.LongRunningOperationStatus statusToWaitFor) {
+        return innerPoller.waitUntil(timeout, statusToWaitFor);
+    }
+
+    @Override
+    public U getFinalResult() {
+        return innerPoller.getFinalResult();
+    }
+
+    @Override
+    public U getFinalResult(Duration timeout) {
+        return innerPoller.getFinalResult(timeout);
+    }
+
+    @Override
+    public void cancelOperation() {
+        innerPoller.cancelOperation();
+    }
+
+    @Override
+    public SyncPoller<PollResult<T>, U> setPollInterval(Duration pollInterval) {
+        innerPoller.setPollInterval(pollInterval);
+        return this;
+    }
+
+    @Override
+    public String serializeContinuationToken() {
+        try {
+            PollingContext<PollResult<T>> context = contextAccessor.getContext();
+            PollingState pollingState = PollingState.from(serializerAdapter, context);
+            return pollingState.toContinuationToken();
+        } catch (Exception e) {
+            throw LOGGER.logExceptionAsError(new RuntimeException("Failed to serialize continuation token. "
+                + "The poller may not have been started or the polling state may be unavailable.", e));
+        }
+    }
+}

sdk/core/azure-core-management/src/main/java/com/azure/core/management/polling/PollerResumeContext.java

@@ -0,0 +1,123 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+package com.azure.core.management.polling;
+
+import com.azure.core.http.HttpPipeline;
+import com.azure.core.util.Context;
+import com.azure.core.util.serializer.SerializerAdapter;
+
+import java.lang.reflect.Type;
+import java.time.Duration;
+import java.util.Objects;
+
+/**
+ * Context object that holds the infrastructure components needed to resume a SyncPoller from a continuation token.
+ * This class simplifies the API for resuming pollers, especially when calling from languages like Scala,
+ * by bundling related parameters together.
+ *
+ * <p><strong>Example usage:</strong></p>
+ * <pre>{@code
+ * // Create a reusable context
+ * PollerResumeContext context = new PollerResumeContext(
+ *     manager.serviceClient().getSerializerAdapter(),
+ *     manager.serviceClient().getHttpPipeline(),
+ *     Duration.ofSeconds(30)
+ * );
+ *
+ * // Resume from token with simpler API
+ * SyncPoller<?, ?> poller = context.resumeFromToken(
+ *     token,
+ *     ServerInner.class,
+ *     ServerInner.class
+ * );
+ * }</pre>
+ */
+public final class PollerResumeContext {
+    private final SerializerAdapter serializerAdapter;
+    private final HttpPipeline httpPipeline;
+    private final Duration defaultPollDuration;
+    private final Context context;
+
+    /**
+     * Creates a PollerResumeContext with the specified infrastructure components.
+     *
+     * @param serializerAdapter The serializer for encoding/decoding.
+     * @param httpPipeline The HTTP pipeline for making requests.
+     * @param defaultPollDuration The default poll interval.
+     */
+    public PollerResumeContext(SerializerAdapter serializerAdapter, HttpPipeline httpPipeline,
+        Duration defaultPollDuration) {
+        this(serializerAdapter, httpPipeline, defaultPollDuration, Context.NONE);
+    }
+
+    /**
+     * Creates a PollerResumeContext with the specified infrastructure components and context.
+     *
+     * @param serializerAdapter The serializer for encoding/decoding.
+     * @param httpPipeline The HTTP pipeline for making requests.
+     * @param defaultPollDuration The default poll interval.
+     * @param context The context for requests.
+     */
+    public PollerResumeContext(SerializerAdapter serializerAdapter, HttpPipeline httpPipeline,
+        Duration defaultPollDuration, Context context) {
+        this.serializerAdapter = Objects.requireNonNull(serializerAdapter, "'serializerAdapter' cannot be null.");
+        this.httpPipeline = Objects.requireNonNull(httpPipeline, "'httpPipeline' cannot be null.");
+        this.defaultPollDuration
+            = Objects.requireNonNull(defaultPollDuration, "'defaultPollDuration' cannot be null.");
+        this.context = Objects.requireNonNull(context, "'context' cannot be null.");
+    }
+
+    /**
+     * Resumes a SyncPoller from a continuation token using this context.
+     *
+     * @param continuationToken The continuation token.
+     * @param pollResultType The type of the poll result.
+     * @param finalResultType The type of the final result.
+     * @param <T> The type of poll result.
+     * @param <U> The type of final result.
+     * @return A SyncPoller that resumes polling from the token.
+     */
+    public <T, U> com.azure.core.util.polling.SyncPoller<PollResult<T>, U> resumeFromToken(String continuationToken,
+        Type pollResultType, Type finalResultType) {
+        return SyncPollerFactory.resumeFromToken(continuationToken, serializerAdapter, httpPipeline, pollResultType,
+            finalResultType, defaultPollDuration, context);
+    }
+
+    /**
+     * Gets the serializer adapter.
+     *
+     * @return The serializer adapter.
+     */
+    public SerializerAdapter getSerializerAdapter() {
+        return serializerAdapter;
+    }
+
+    /**
+     * Gets the HTTP pipeline.
+     *
+     * @return The HTTP pipeline.
+     */
+    public HttpPipeline getHttpPipeline() {
+        return httpPipeline;
+    }
+
+    /**
+     * Gets the default poll duration.
+     *
+     * @return The default poll duration.
+     */
+    public Duration getDefaultPollDuration() {
+        return defaultPollDuration;
+    }
+
+    /**
+     * Gets the context.
+     *
+     * @return The context.
+     */
+    public Context getContext() {
+        return context;
+    }
+}
+