Handle spot instance interruptions with SMQ

Workflow overview

Prerequisites

1. Create an AccessKey

   Use the AccessKey of a RAM user with minimum required permissions instead of the root account. See Create an AccessKey.

2. Grant permissions to the RAM user

   Grant the RAM user the following permission to consume messages from Simple Message Queue (MNS):

   Cloud Products	Grant permissions
   Simple Message Queue (SMQ, formerly MNS)	AliyunMNSFullAccess

3. Configure access credentials and endpoint

   The sample code reads access credentials and the endpoint from environment variables.

   • Access credentials: see Configure environment variables on Linux, macOS, and Windows.

   • Endpoint: see Configure the endpoint.

4. Install the SMQ SDK

   Install the SMQ Java SDK by adding a Maven dependency. For other methods, see Install the SMQ SDK.

   Add the following Maven dependency:

   <dependencies>
       <!-- Alibaba Cloud Simple Message Queue SDK -->
       <dependency>
         <groupId>com.aliyun.mns</groupId>
         <artifactId>aliyun-sdk-mns</artifactId>
         <version>1.2.0</version>
       </dependency>
   </dependencies>

Procedure

1. Create a Simple Message Queue

   Create a queue to receive spot instance interruption notifications.

   1. Log on to the Simple Message Queue (formerly MNS) console. In the left-side navigation pane, choose Queue Model > Queues.

   2. On the top menu bar, select a region. On the Queues page, click Create Queue.

   3. In the Create Queue panel, configure the parameters and click OK.

      The configuration parameters include:

      • Name: Queue name, up to 120 characters.

      • Maximum Message Size: Default: 64 KB.

      • Long Polling Wait Time: Default: 15 seconds.

      • Message Visibility Timeout: Default: 30 seconds.

      • Message Retention Period: Default: 4 days.

      • Message Receive Delay: Default: 0 seconds.

      • Enable Logging: Default: Disabled.

      • Dead-Letter Policy: Default: Disabled.

2. Create a subscription policy

   Cloud Monitor sends spot instance interruption notifications through the push channel specified in the subscription policy.

   1. Log on to the Cloud Monitor console. In the left-side navigation pane, choose Event Hub > Event Subscriptions.

   2. On the Subscription Policies tab, click Create Subscription Policy. Click Next and configure the parameters.

      Only key parameters are described here. For other parameters, see Manage event subscriptions (recommended).

      • Subscription Type: Select System Events.

      • Subscription Scope: Configure the following parameters:

        • Product: Select Elastic Compute Service.

        • Event Type: Select Status Notification.

        • Event Name: Select Spot Instance Interruption Notification.

        • Event Level: Select Warning.

      • Push and Integration: Click + Add Channel > Add Channel, select the queue created in Step 1, and complete the remaining fields. See Manage Push Channels.

3. Simulate an interruption event

   Use Test Event Subscription to simulate an interruption event, because spot instance interruptions are triggered passively and difficult to reproduce during development.

   1. On the Subscription Policies tab, click Test Event Subscription.

   2. In the Create Event Debugging panel, set Service to ECS and Name to Spot Instance Interruption Notification.

      The system generates JSON debugging content. Replace the resource-related fields with your spot instance information.

      • Replace Alibaba Cloud account UID with your account UID.

      • Replace <resource-id> and <instanceId> with your spot instance ID.

      • Replace <region-id> with your spot instance region ID.

        {
            "product": "ECS",
            "resourceId": "acs:ecs:cn-shanghai:Alibaba Cloud account UID:instance/<resource-id>",
            "level": "WARN",
            "instanceName": "instanceName",
            "regionId": "<region-id>",
            "groupId": "0",
            "name": "Instance:PreemptibleInstanceInterruption",
            "content": {
                "instanceId": "<instanceId>",
                "instanceName": "wor***b73",
                "action": "delete"
            },
            "status": "Normal"
        }

   3. Click OK. The system displays The operation is successful., and Cloud Monitor sends a test notification to your queue.

4. Pull and respond to messages

   Pull interruption notifications from the queue and respond with custom logic. The following sample code demonstrates handling an interruption event with an image grayscale conversion program:

   1. Use a thread task to simulate the image conversion program.

      import com.aliyun.mns.client.CloudAccount;
      import com.aliyun.mns.client.CloudQueue;
      import com.aliyun.mns.client.MNSClient;
      import com.aliyun.mns.common.utils.ServiceSettings;
      import com.aliyun.mns.model.Message;
      import org.json.JSONObject;
      
      import javax.imageio.ImageIO;
      import java.awt.image.BufferedImage;
      import java.io.File;
      import java.util.Base64;
      import java.util.concurrent.atomic.AtomicBoolean;
      
      /**
       * Interruptible image processor that supports grayscale conversion.
       * Uses atomic variables and thread interruption flags for detection.
       * Features:
       * 1. Chunked processing with automatic progress saving
       * 2. Immediate response to spot instance interruption
       * 3. Partial result file generation after interruption
       */
      public class InterruptibleImageProcessor implements Runnable {
          /**
           * Thread-safe state control using an atomic Boolean
           */
          private final AtomicBoolean running = new AtomicBoolean(true);
          /**
           * Stores image data being processed
           */
          private BufferedImage processedImage;
          /**
           * Processing progress percentage (0–100)
           */
          private int progress;
          /**
           * Thread entry point
           * **Interruption logic**:
           * 1. Save current progress after catching InterruptedException
           * 2. Restore thread interruption status (preserve semantics)
           */
          @Override
          public void run() {
              try {
                  convertToGrayScale(new File("input.jpg"), new File("output.jpg"));
                  // Simulate post-processing delay
                  Thread.sleep(5000); 
                  System.out.println("Image processing completed");
              } catch (InterruptedException e) {
                  System.out.println("Processing interrupted. Progress saved at " + progress + "%");
                  saveProgress(new File("partial_output.jpg"));
                  Thread.currentThread().interrupt(); // Restore interruption status
              } catch (Exception e) {
                  System.err.println("Processing error: " + e.getMessage());
              }
          }
      
          /**
           * External interruption trigger method
           * **Coordination mechanism**:
           * Works with thread interruption flag for dual detection
           */
          public void stop() {
              running.set(false);
          }
      }
      

   2. Image grayscale conversion method.

      /**
       * Convert input image to grayscale and save
       * @param inputFile Source image file object
       * @param outputFile Output file object
       * @throws Exception Includes I/O and interruption exceptions
       *
       * **Algorithm**:
       * Weighted average grayscale conversion matching human luminance perception:
       * Gray = 0.30*R + 0.59*G + 0.11*B
       * Reference: ITU-R BT.601 standard
       */
      public void convertToGrayScale(File inputFile, File outputFile) throws Exception {
          // Read source image
          BufferedImage original = ImageIO.read(inputFile);
          int width = original.getWidth();
          int height = original.getHeight();
          // Create grayscale image buffer
          processedImage = new BufferedImage(width, height, BufferedImage.TYPE_BYTE_GRAY);
      
          // Process in chunks to support progress saving
          for (int y = 0; y < height && running.get(); y++) {
              // Process pixel by pixel
              for (int x = 0; x < width; x++) {
                  // First interruption check: thread interruption flag
                  if (Thread.interrupted()) {
                      throw new InterruptedException("Image processing interrupted");
                  }
                  /* Core grayscale algorithm */
                  int rgb = original.getRGB(x, y);
                  // Extract RGB channels (ARGB format)
                  // Red channel
                  int r = (rgb >> 16) & 0xFF;
                  // Green channel
                  int g = (rgb >> 8) & 0xFF;
                  // Blue channel
                  int b = rgb & 0xFF;
                  // Calculate grayscale value (weighted average)
                  int gray = (int)(0.3 * r + 0.59 * g + 0.11 * b);
                  // Reconstruct RGB (grayscale value replicated across channels)
                  processedImage.setRGB(x, y, (gray << 16) | (gray << 8) | gray);
                  // Update progress percentage (watch for integer division)
                  progress = (y * width + x) * 100 / (width * height);
              }
              // Auto-save progress every 50 rows (checkpoint mechanism)
              if (y % 50 == 0) {
                  saveProgress(outputFile);
              }
          }
          // Save final result
          ImageIO.write(processedImage, "jpg", outputFile);
      }

   3. Save image processing progress.

      /**
       * Save processing progress to specified file
       * @param outputFile Output file object
       *
       * **Note**:
       * 1. Uses silent failure to avoid interrupting save process
       * 2. Generates temporary file named partial_output.jpg
       */
      private void saveProgress(File outputFile) {
          try {
              // Use temporary filename to avoid overwriting final file
              ImageIO.write(processedImage, "jpg", new File("partial_output.jpg"));
          } catch (Exception e) {
              System.err.println("Auto-save failed: " + e.getMessage());
          }
      }

   4. Test the response handling.

      Test the image processor to detect and respond to a spot instance interruption while running.

      /**
       * Main method (for testing)
       * **Test scenario**:
       * 1. Start processing thread
       * 2. Pull interruption notification message
       * 3. Wait for thread termination
       */
      public static void main(String[] args) throws InterruptedException {
          // Initialize MNS client
          CloudAccount account = new CloudAccount(
                  ServiceSettings.getMNSAccessKeyId(),
                  ServiceSettings.getMNSAccessKeySecret(),
                  ServiceSettings.getMNSAccountEndpoint());
          MNSClient client = account.getMNSClient();
          // Check if message matches spot instance interruption event
          boolean isMatch = false;
          // Start image processor
          InterruptibleImageProcessor processor = new InterruptibleImageProcessor();
          Thread processThread = new Thread(processor);
          processThread.start();
          try{
              // Pull message from queue
              CloudQueue queue = client.getQueueRef("spot-interruption");
              Message popMsg = queue.popMessage();
              if (popMsg != null){
                  // Message body is Base64-encoded by default
                  System.out.println("message body: " + popMsg.getMessageBodyAsRawString());
                  // Base64 decode
                  byte[] decodedBytes = Base64.getDecoder().decode(popMsg.getMessageBodyAsRawString());
                  String decodedString = new String(decodedBytes);
                  System.out.println("message content: " + decodedString);
                  // Parse JSON string
                  JSONObject json = new JSONObject(decodedString);
                  // Get event name from "name" field
                  String name = json.getString("name");
                  isMatch = "Instance:PreemptibleInstanceInterruption".equals(name);
                  // Handle spot instance interruption event
                  if(isMatch){
                      System.out.println("Spot instance will be interrupted and reclaimed");
                      // Stop image processor
                      processor.stop();
                      processThread.interrupt();
                      System.out.println("Program terminated");
                      processThread.join();
                      // Delete message
                      queue.deleteMessage(popMsg.getReceiptHandle());
                  }
              }
          }catch (Exception e){
              System.out.println("Unknown exception happened!");
              e.printStackTrace();
          }
          client.close();
      }

   Complete sample code:

   import com.aliyun.mns.client.CloudAccount;
   import com.aliyun.mns.client.CloudQueue;
   import com.aliyun.mns.client.MNSClient;
   import com.aliyun.mns.common.utils.ServiceSettings;
   import com.aliyun.mns.model.Message;
   import org.json.JSONObject;
   import javax.imageio.ImageIO;
   import java.awt.image.BufferedImage;
   import java.io.File;
   import java.util.Base64;
   import java.util.concurrent.atomic.AtomicBoolean;
   
   /**
    * Interruptible image processor that supports grayscale conversion.
    * Uses atomic variables and thread interruption flags for detection.
    * Features:
    * 1. Chunked processing with automatic progress saving
    * 2. Immediate response to spot instance interruption
    * 3. Partial result file generation after interruption
    */
   public class InterruptibleImageProcessor implements Runnable {
       /**
        * Thread-safe state control using an atomic Boolean
        */
       private final AtomicBoolean running = new AtomicBoolean(true);
       /**
        * Stores image data being processed
        */
       private BufferedImage processedImage;
       /**
        * Processing progress percentage (0–100)
        */
       private int progress;
   
       /**
        * Convert input image to grayscale and save
        * @param inputFile Source image file object
        * @param outputFile Output file object
        * @throws Exception Includes I/O and interruption exceptions
        *
        * **Algorithm**:
        * Weighted average grayscale conversion matching human luminance perception:
        * Gray = 0.30*R + 0.59*G + 0.11*B
        * Reference: ITU-R BT.601 standard
        */
       public void convertToGrayScale(File inputFile, File outputFile) throws Exception {
           // Read source image
           BufferedImage original = ImageIO.read(inputFile);
           int width = original.getWidth();
           int height = original.getHeight();
           // Create grayscale image buffer
           processedImage = new BufferedImage(width, height, BufferedImage.TYPE_BYTE_GRAY);
   
           // Process in chunks to support progress saving
           for (int y = 0; y < height && running.get(); y++) {
               // Process pixel by pixel
               for (int x = 0; x < width; x++) {
                   // First interruption check: thread interruption flag
                   if (Thread.interrupted()) {
                       throw new InterruptedException("Image processing interrupted");
                   }
                   /* Core grayscale algorithm */
                   int rgb = original.getRGB(x, y);
                   // Extract RGB channels (ARGB format)
                   // Red channel
                   int r = (rgb >> 16) & 0xFF;
                   // Green channel
                   int g = (rgb >> 8) & 0xFF;
                   // Blue channel
                   int b = rgb & 0xFF;
                   // Calculate grayscale value (weighted average)
                   int gray = (int)(0.3 * r + 0.59 * g + 0.11 * b);
                   // Reconstruct RGB (grayscale value replicated across channels)
                   processedImage.setRGB(x, y, (gray << 16) | (gray << 8) | gray);
                   // Update progress percentage (watch for integer division)
                   progress = (y * width + x) * 100 / (width * height);
               }
               // Auto-save progress every 50 rows (checkpoint mechanism)
               if (y % 50 == 0) {
                   saveProgress(outputFile);
               }
           }
           // Save final result
           ImageIO.write(processedImage, "jpg", outputFile);
       }
   
       /**
        * Save processing progress to specified file
        * @param outputFile Output file object
        *
        * **Note**:
        * 1. Uses silent failure to avoid interrupting save process
        * 2. Generates temporary file named partial_output.jpg
        */
       private void saveProgress(File outputFile) {
           try {
               // Use temporary filename to avoid overwriting final file
               ImageIO.write(processedImage, "jpg", new File("partial_output.jpg"));
           } catch (Exception e) {
               System.err.println("Auto-save failed: " + e.getMessage());
           }
       }
   
       /**
        * Thread entry point
        * **Interruption logic**:
        * 1. Save current progress after catching InterruptedException
        * 2. Restore thread interruption status (preserve semantics)
        */
       @Override
       public void run() {
           try {
               convertToGrayScale(new File("/Users/shaoberlin/Desktop/idea_workspace/aliyun/src/main/resources/input.jpg"), new File("output.jpg"));
               // Simulate post-processing delay
               Thread.sleep(5000); 
               System.out.println("Image processing completed");
           } catch (InterruptedException e) {
               System.out.println("Processing interrupted. Progress saved at " + progress + "%");
               saveProgress(new File("partial_output.jpg"));
               Thread.currentThread().interrupt(); // Restore interruption status
           } catch (Exception e) {
               System.err.println("Processing error: " + e.getMessage());
           }
       }
   
       /**
        * External interruption trigger method
        * **Coordination mechanism**:
        * Works with thread interruption flag for dual detection
        */
       public void stop() {
           running.set(false);
       }
   
       /**
        * Main method (for testing)
        * **Test scenario**:
        * 1. Start processing thread
        * 2. Trigger interruption after 5000 ms
        * 3. Wait for thread termination
        */
       public static void main(String[] args) throws InterruptedException {
           // Initialize MNS client
           CloudAccount account = new CloudAccount(
                   ServiceSettings.getMNSAccessKeyId(),
                   ServiceSettings.getMNSAccessKeySecret(),
                   ServiceSettings.getMNSAccountEndpoint());
           MNSClient client = account.getMNSClient();
           // Check if message matches spot instance interruption event
           boolean isMatch = false;
           // Start image processor
           InterruptibleImageProcessor processor = new InterruptibleImageProcessor();
           Thread processThread = new Thread(processor);
           processThread.start();
           try{
               // Pull message from queue
               CloudQueue queue = client.getQueueRef("spot-interruption");
               Message popMsg = queue.popMessage();
               if (popMsg != null){
                   // Message body is Base64-encoded by default
                   System.out.println("message body: " + popMsg.getMessageBodyAsRawString());
                   // Base64 decode
                   byte[] decodedBytes = Base64.getDecoder().decode(popMsg.getMessageBodyAsRawString());
                   String decodedString = new String(decodedBytes);
                   System.out.println("message content: " + decodedString);
                   // Parse JSON string
                   JSONObject json = new JSONObject(decodedString);
                   // Get event name from "name" field
                   String name = json.getString("name");
                   isMatch = "Instance:PreemptibleInstanceInterruption".equals(name);
                   // Handle spot instance interruption event
                   if(isMatch){
                       System.out.println("Spot instance will be interrupted and reclaimed");
                       // Stop image processor
                       processor.stop();
                       processThread.interrupt();
                       System.out.println("Program terminated");
                       processThread.join();
                       // Delete message
                       queue.deleteMessage(popMsg.getReceiptHandle());
                   }
               }
           }catch (Exception e){
               System.out.println("Unknown exception happened!");
               e.printStackTrace();
           }
           client.close();
       }
   }