Spring Cloud Azure

spring:
  cloud:
    azure:
      profile:
        tenant-id: ${AZURE_TENANT_ID}
      credential:
        client-id: ${AZURE_CLIENT_ID}
        client-secret: ${AZURE_CLIENT_SECRET}
      keyvault:
        secret:
          property-source-enabled: true
          endpoint: ${AZURE_KEYVAULT_ENDPOINT}

@SpringBootApplication
public class KeyVaultSample implements CommandLineRunner {

    @Value("${your-property-name}")
    private String mySecretProperty;

    public static void main(String[] args) {
        SpringApplication.run(KeyVaultSample.class, args);
    }

    @Override
    public void run(String... args) {
        System.out.println("property your-property-name value is: " + mySecretProperty);
    }
}

spring:
  cloud:
    azure:
      keyvault:
        secret:
          property-source-enabled: true
          property-sources:
            -
              name: key-vault-1
              endpoint: ${ENDPOINT_1}
              profile:
                tenant-id: ${AZURE_TENANT_ID_1}
              credential:
                client-id: ${AZURE_CLIENT_ID_1}
                client-secret: ${AZURE_CLIENT_SECRET_1}
            -
              name: key-vault-2
              endpoint: ${ENDPOINT_2}
              profile:
                tenant-id: ${AZURE_TENANT_ID_2}
              credential:
                client-id: ${AZURE_CLIENT_ID_2}
                client-secret: ${AZURE_CLIENT_SECRET_2}

@SpringBootApplication
public class SampleApplication implements CommandLineRunner {

    @Value("${secret-name-in-key-vault-1}")
    private String secretNameInKeyVault1;
    @Value("${secret-name-in-key-vault-2}")
    private String secretNameInKeyVault2;
    @Value("${secret-name-in-key-vault-both}")
    private String secretNameInKeyVaultBoth;

    public static void main(String[] args) {
        SpringApplication.run(SampleApplication.class, args);
    }

    public void run(String[] args) {
        System.out.println("secretNameInKeyVault1: " + secretNameInKeyVault1);
        System.out.println("secretNameInKeyVault2: " + secretNameInKeyVault2);
        System.out.println("secretNameInKeyVaultBoth: " + secretNameInKeyVaultBoth);
    }

}

spring:
  cloud:
    azure:
      active-directory:
        base-uri: https://login.partner.microsoftonline.cn
        graph-base-uri: https://microsoftgraph.chinacloudapi.cn

spring:
  cloud:
    azure:
      active-directory:
        user-group:
          allowed-group-names: group1_name_1, group2_name_2
          # 1. If allowed-group-ids == all, then all group id will take effect.
          # 2. If "all" is used, we should not configure other group ids.
          # 3. "all" is only supported for allowed-group-ids, not supported for allowed-group-names.
          allowed-group-ids: group_id_1, group_id_2

@EnableWebSecurity
@EnableGlobalMethodSecurity(prePostEnabled == true)
public class AADOAuth2LoginSecurityConfig extends AADWebSecurityConfigurerAdapter {

    /**
     * Add configuration logic as needed.
     */
    @Override
    protected void configure(HttpSecurity http) throws Exception {
        super.configure(http);
        http.authorizeRequests()
                .anyRequest().authenticated();
        // Do some custom configuration
    }
}

@Controller
public class RoleController {
    @GetMapping("group1")
    @ResponseBody
    @PreAuthorize("hasRole('ROLE_group1')")
    public String group1() {
        return "group1 message";
    }

    @GetMapping("group2")
    @ResponseBody
    @PreAuthorize("hasRole('ROLE_group2')")
    public String group2() {
        return "group2 message";
    }

    @GetMapping("group1Id")
    @ResponseBody
    @PreAuthorize("hasRole('ROLE_<group1-id>')")
    public String group1Id() {
        return "group1Id message";
    }

    @GetMapping("group2Id")
    @ResponseBody
    @PreAuthorize("hasRole('ROLE_<group2-id>')")
    public String group2Id() {
        return "group2Id message";
    }
}

spring:
  cloud:
    azure:
      active-directory:
        authorization-clients:
          graph:
            scopes: https://graph.microsoft.com/Analytics.Read, email
          arm: # client registration id
            on-demand: true  # means incremental consent
            scopes: https://management.core.windows.net/user_impersonation

spring:
  cloud:
    azure:
      active-directory:
        authorization-clients:
          webapiC:                          # When authorization-grant-type is null, on behalf of flow is used by default
            authorization-grant-type: client_credentials
            scopes:
                - <Web-API-C-app-id-url>/.default

<dependencies>
    <dependency>
        <groupId>com.azure.spring</groupId>
        <artifactId>spring-cloud-azure-starter-active-directory</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-oauth2-client</artifactId>
    </dependency>
</dependencies>

spring:
  cloud:
    azure:
      active-directory:
        tenant-id: ${AZURE_TENANT_ID}
        client-id: ${AZURE_CLIENT_ID}
        client-secret: ${AZURE_CLIENT_SECRET}

<dependencies>
    <dependency>
        <groupId>com.azure.spring</groupId>
        <artifactId>spring-cloud-azure-starter-active-directory</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-oauth2-client</artifactId>
    </dependency>
</dependencies>

spring:
  cloud:
    azure:
      active-directory:
        tenant-id: ${AZURE_TENANT_ID}
        client-id: ${AZURE_CLIENT_ID}
        client-secret: ${AZURE_CLIENT_SECRET}
        authorization-clients:
          graph:
            scopes: https://graph.microsoft.com/Analytics.Read, email

<dependencies>
    <dependency>
        <groupId>com.azure.spring</groupId>
        <artifactId>spring-cloud-azure-starter-active-directory</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-oauth2-resource-server</artifactId>
    </dependency>
</dependencies>

<dependencies>
    <dependency>
        <groupId>com.azure.spring</groupId>
        <artifactId>spring-cloud-azure-starter-active-directory</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-oauth2-resource-server</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-oauth2-client</artifactId>
    </dependency>
</dependencies>

spring:
  cloud:
    azure:
      active-directory:
        tenant-id: ${AZURE_TENANT_ID}
        client-id: ${AZURE_CLIENT_ID}
        client-secret: ${AZURE_CLIENT_SECRET}
        app-id-uri: ${WEB_API_ID_URI}
        authorization-clients:
          graph:
            scopes:
              - https://graph.microsoft.com/User.Read

public class SampleController {
    @PreAuthorize("hasAuthority('SCOPE_Obo.Graph.Read')")
    @GetMapping("call-graph")
    public String callGraph(@RegisteredOAuth2AuthorizedClient("graph") OAuth2AuthorizedClient graph) {
        return callMicrosoftGraphMeEndpoint(graph);
    }
}

<dependencies>
    <dependency>
        <groupId>com.azure.spring</groupId>
        <artifactId>spring-cloud-azure-starter-active-directory</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-oauth2-resource-server</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-oauth2-client</artifactId>
    </dependency>
</dependencies>

spring:
  cloud:
    azure:
      active-directory:
        tenant-id: ${AZURE_TENANT_ID}
        client-id: ${AZURE_CLIENT_ID}
        client-secret: ${AZURE_CLIENT_SECRET}
        app-id-uri: ${WEB_API_ID_URI}
        application-type: web_application_and_resource_server  # This is required.
        authorization-clients:
          graph:
            authorizationGrantType: authorization_code # This is required.
            scopes:
              - https://graph.microsoft.com/User.Read
              - https://graph.microsoft.com/Directory.Read.All

@EnableWebSecurity
@EnableGlobalMethodSecurity(prePostEnabled == true)
public class AADWebApplicationAndResourceServerConfig {

    @Order(1)
    @Configuration
    public static class ApiWebSecurityConfigurationAdapter extends AADResourceServerWebSecurityConfigurerAdapter {
        protected void configure(HttpSecurity http) throws Exception {
            super.configure(http);
            // All the paths that match `/api/**`(configurable) work as `Resource Server`, other paths work as `Web application`.
            http.antMatcher("/api/**")
                .authorizeRequests().anyRequest().authenticated();
        }
    }

    @Configuration
    public static class HtmlWebSecurityConfigurerAdapter extends AADWebSecurityConfigurerAdapter {

        @Override
        protected void configure(HttpSecurity http) throws Exception {
            super.configure(http);
            // @formatter:off
            http.authorizeRequests()
                    .antMatchers("/login").permitAll()
                    .anyRequest().authenticated();
            // @formatter:on
        }
    }
}

<dependencies>
    <dependency>
        <groupId>com.azure.spring</groupId>
        <artifactId>azure-spring-boot-starter-active-directory-b2c</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-thymeleaf</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-security</artifactId>
    </dependency>
    <dependency>
        <groupId>org.thymeleaf.extras</groupId>
        <artifactId>thymeleaf-extras-springsecurity5</artifactId>
    </dependency>
</dependencies>

spring:
  cloud:
    azure:
      active-directory:
        b2c:
          authenticate-additional-parameters:
            domain_hint: xxxxxxxxx         # optional
            login_hint: xxxxxxxxx          # optional
            prompt: [login,none,consent]   # optional
          base-uri: ${BASE_URI}
          client-id: ${WEBAPP_AZURE_CLIENT_ID}
          client-secret: ${WEBAPP_AZURE_CLIENT_SECRET}
          login-flow: ${LOGIN_USER_FLOW_KEY}               # default to sign-up-or-sign-in, will look up the user-flows map with provided key.
          logout-success-url: ${LOGOUT_SUCCESS_URL}
          user-flows:
            ${YOUR_USER_FLOW_KEY}: ${USER_FLOW_NAME}
          user-name-attribute-name: ${USER_NAME_ATTRIBUTE_NAME}

@EnableWebSecurity
public class WebSecurityConfiguration extends WebSecurityConfigurerAdapter {

    private final AADB2COidcLoginConfigurer configurer;

    public WebSecurityConfiguration(AADB2COidcLoginConfigurer configurer) {
        this.configurer == configurer;
    }

    @Override
    protected void configure(HttpSecurity http) throws Exception {
        // @formatter:off
        http.authorizeRequests()
                .anyRequest().authenticated()
                .and()
            .apply(configurer);
        // @formatter:off
    }
}

{
  "allowedMemberTypes": [
    "Application"
  ],
  "description": "WebApiA.SampleScope",
  "displayName": "WebApiA.SampleScope",
  "id": "04989db0-3efe-4db6-b716-ae378517d2b7",
  "isEnabled": true,
  "value": "WebApiA.SampleScope"
}

<dependency>
  <groupId>org.springframework.boot</groupId>
  <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

spring:
  cloud:
    azure:
      active-directory:
        b2c:
          base-uri: ${BASE_URI}             # Such as: https://xxxxb2c.b2clogin.com
          tenant-id: ${AZURE_TENANT_ID}
          authorization-clients:
            ${RESOURCE_SERVER_A_NAME}:
              authorization-grant-type: client_credentials
              scopes: ${WEB_API_A_APP_ID_URL}/.default

public class SampleConfiguration {
        @Bean
        public WebClient webClient(OAuth2AuthorizedClientManager oAuth2AuthorizedClientManager) {
                ServletOAuth2AuthorizedClientExchangeFilterFunction function =
                        new ServletOAuth2AuthorizedClientExchangeFilterFunction(oAuth2AuthorizedClientManager);
                return WebClient.builder()
                                                .apply(function.oauth2Configuration())
                                                .build();
        }
}

<dependencies>
    <dependency>
        <groupId>com.azure.spring</groupId>
        <artifactId>azure-spring-boot-starter-active-directory-b2c</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
</dependencies>

spring:
  cloud:
    azure:
      active-directory:
        b2c:
          base-uri: ${BASE_URI}             # Such as: https://xxxxb2c.b2clogin.com
          tenant-id: ${AZURE_TENANT_ID}
          app-id-uri: ${APP_ID_URI}         # If you are using v1.0 token, please configure app-id-uri for `aud` verification
          client-id: ${AZURE_CLIENT_ID}           # If you are using v2.0 token, please configure client-id for `aud` verification

@EnableWebSecurity
@EnableGlobalMethodSecurity(prePostEnabled == true)
public class ResourceServerConfiguration extends WebSecurityConfigurerAdapter {

    @Override
    protected void configure(HttpSecurity http) throws Exception {
        http.authorizeRequests((requests) -> requests.anyRequest().authenticated())
            .oauth2ResourceServer()
            .jwt()
            .jwtAuthenticationConverter(new AADJwtBearerTokenAuthenticationConverter());
    }
}

{
    "allowedMemberTypes": [
        "Application"
    ],
    "description": "WebApiB.SampleScope",
    "displayName": "WebApiB.SampleScope",
    "id": "04989db0-3efe-4db6-b716-ae378517d2b7",
    "isEnabled": true,
    "lang": null,
    "origin": "Application",
    "value": "WebApiB.SampleScope"
}

<dependency>
 <groupId>org.springframework.boot</groupId>
 <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

spring:
  cloud:
    azure:
      active-directory:
        b2c:
          client-secret: ${WEB_API_A_AZURE_CLIENT_SECRET}
          authorization-clients:
            ${RESOURCE_SERVER_B_NAME}:
              authorization-grant-type: client_credentials
              scopes: ${WEB_API_B_APP_ID_URL}/.default

public class SampleController {
        /**
         * webApiB resource api for other web application
         * @return test content
         */
        @PreAuthorize("hasAuthority('APPROLE_WebApiB.SampleScope')")
        @GetMapping("/webApiB/sample")
        public String webApiBSample() {
                LOGGER.info("Call webApiBSample()");
                return "Request '/webApiB/sample'(WebApi B) returned successfully.";
        }
}

spring:
  cloud:
    azure:
      eventhubs:
        connection-string: ${AZURE_SERVICE_BUS_CONNECTION_STRING}

spring:
  cloud:
    azure:
      credential:
        managed-identity-client-id: ${AZURE_CLIENT_ID}
      profile:
        tenant-id: ${AZURE_TENANT_ID}
#     Uncomment below configurations if you want to enable auto creating resources.
#        subscription-id: ${AZURE_SUBSCRIPTION_ID}
#        cloud: Azure
#      resource:
#        region: [region]

      eventhubs:
        namespace: ${AZURE_SERVICE_BUS_NAMESPACE}

spring:
  cloud:
    azure:
      credential:
        client-id: ${AZURE_CLIENT_ID}
        client-secret: ${AZURE_CLIENT_SECRET}
      profile:
        tenant-id: ${AZURE_TENANT_ID}
#     Uncomment below configurations if you want to enable auto creating resources.
#        subscription-id: ${AZURE_SUBSCRIPTION_ID}
#        cloud: Azure
#      resource:
#        region: [region]

      eventhubs:
        namespace: ${AZURE_SERVICE_BUS_NAMESPACE}

private static final String OUTPUT_CHANNEL = "output";
private static final String EVENTHUB_NAME = "eh1";

@Bean
@ServiceActivator(inputChannel = OUTPUT_CHANNEL)
public MessageHandler messageSender(EventHubsTemplate queueOperation) {
    DefaultMessageHandler handler = new DefaultMessageHandler(EVENTHUB_NAME, queueOperation);
    handler.setSendCallback(new ListenableFutureCallback<Void>() {
        @Override
        public void onSuccess(Void result) {
            LOGGER.info("Message was sent successfully.");
        }
        @Override
        public void onFailure(Throwable ex) {
            LOGGER.error("There was an error sending the message.", ex);
        }
    });
    return handler;
}

@Autowired
EventHubOutboundGateway messagingGateway;

@MessagingGateway(defaultRequestChannel = OUTPUT_CHANNEL)
public interface EventHubOutboundGateway {
    void send(String text);
}

this.messagingGateway.send(message);

private static final String INPUT_CHANNEL = "input";
private static final String EVENTHUB_NAME = "eh1";
private static final String CONSUMER_GROUP = "$Default";

@Bean
public MessageChannel input() {
    return new DirectChannel();
}

@Bean
public EventHubsInboundChannelAdapter messageChannelAdapter(
        @Qualifier(INPUT_CHANNEL) MessageChannel inputChannel,
        EventHubsProcessorContainer processorContainer) {
    CheckpointConfig config = new CheckpointConfig(CheckpointMode.MANUAL);

    EventHubsInboundChannelAdapter adapter =
            new EventHubsInboundChannelAdapter(processorContainer, EVENTHUB_NAME,
                    CONSUMER_GROUP, config);
    adapter.setOutputChannel(inputChannel);
    return adapter;
}

@ServiceActivator(inputChannel = INPUT_CHANNEL)
public void messageReceiver(byte[] payload, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) {
    String message = new String(payload);
    LOGGER.info("New message received: '{}'", message);
    checkpointer.success()
            .doOnSuccess(s -> LOGGER.info("Message '{}' successfully checkpointed", message))
            .doOnError(e -> LOGGER.error("Error found", e))
            .subscribe();
}

spring:
  cloud:
    azure:
      servicebus:
        connection-string: ${AZURE_SERVICE_BUS_CONNECTION_STRING}

spring:
  cloud:
    azure:
      credential:
        managed-identity-client-id: ${AZURE_CLIENT_ID}
      profile:
        tenant-id: ${AZURE_TENANT_ID}
#     Uncomment below configurations if you want to enable auto creating resources.
#        subscription-id: ${AZURE_SUBSCRIPTION_ID}
#        cloud: Azure
#      resource:
#        region: [region]

      servicebus:
        namespace: ${AZURE_SERVICE_BUS_NAMESPACE}

spring:
  cloud:
    azure:
      credential:
        client-id: ${AZURE_CLIENT_ID}
        client-secret: ${AZURE_CLIENT_SECRET}
      profile:
        tenant-id: ${AZURE_TENANT_ID}
#     Uncomment below configurations if you want to enable auto creating resources.
#        subscription-id: ${AZURE_SUBSCRIPTION_ID}
#        cloud: Azure
#      resource:
#        region: [region]

      servicebus:
        namespace: ${AZURE_SERVICE_BUS_NAMESPACE}

private static final String OUTPUT_CHANNEL = "queue.output";

@Bean
@ServiceActivator(inputChannel = OUTPUT_CHANNEL)
public MessageHandler queueMessageSender(ServiceBusTemplate serviceBusTemplate) {
    serviceBusTemplate.setDefaultEntityType(ServiceBusEntityType.QUEUE);
    DefaultMessageHandler handler = new DefaultMessageHandler(QUEUE_NAME, serviceBusTemplate);
    handler.setSendCallback(new ListenableFutureCallback<Void>() {
        @Override
        public void onSuccess(Void result) {
            LOGGER.info("Message was sent successfully.");
        }

        @Override
        public void onFailure(Throwable ex) {
            LOGGER.info("There was an error sending the message.");
        }
    });

    return handler;
}

@Autowired
QueueOutboundGateway messagingGateway;

@MessagingGateway(defaultRequestChannel = OUTPUT_CHANNEL)
public interface QueueOutboundGateway {
    void send(String text);
}

this.messagingGateway.send(message);

private static final String INPUT_CHANNEL = "input";

@Bean
public MessageChannel input() {
    return new DirectChannel();
}

private static final String QUEUE_NAME = "queue1";

@Bean
public ServiceBusInboundChannelAdapter queueMessageChannelAdapter(
    @Qualifier(INPUT_CHANNEL) MessageChannel inputChannel, ServiceBusProcessorContainer processorContainer) {
    ServiceBusInboundChannelAdapter adapter = new ServiceBusInboundChannelAdapter(processorContainer, QUEUE_NAME,
        new CheckpointConfig(CheckpointMode.MANUAL));
    adapter.setOutputChannel(inputChannel);
    return adapter;
}

@ServiceActivator(inputChannel = INPUT_CHANNEL)
public void messageReceiver(byte[] payload, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) {
    String message = new String(payload);
    LOGGER.info("New message received: '{}'", message);
    checkpointer.success()
            .doOnSuccess(s -> LOGGER.info("Message '{}' successfully checkpointed", message))
            .doOnError(e -> LOGGER.error("Error found", e))
            .subscribe();
}

spring:
  cloud:
    azure:
      storage:
        queue:
          connection-string: ${AZURE_SERVICE_BUS_CONNECTION_STRING}

spring:
  cloud:
    azure:
      credential:
        managed-identity-client-id: ${AZURE_CLIENT_ID}
      profile:
        tenant-id: ${AZURE_TENANT_ID}
#     Uncomment below configurations if you want to enable auto creating resources.
#        subscription-id: ${AZURE_SUBSCRIPTION_ID}
#        cloud: Azure
#      resource:
#        region: [region]

      storage:
        queue:
          namespace: ${AZURE_SERVICE_BUS_NAMESPACE}

spring:
  cloud:
    azure:
      credential:
        client-id: ${AZURE_CLIENT_ID}
        client-secret: ${AZURE_CLIENT_SECRET}
      profile:
        tenant-id: ${AZURE_TENANT_ID}
#     Uncomment below configurations if you want to enable auto creating resources.
#        subscription-id: ${AZURE_SUBSCRIPTION_ID}
#        cloud: Azure
#      resource:
#        region: [region]

      storage:
        queue:
          namespace: ${AZURE_SERVICE_BUS_NAMESPACE}

private static final String STORAGE_QUEUE_NAME = "example";
private static final String OUTPUT_CHANNEL = "output";

@Bean
@ServiceActivator(inputChannel = OUTPUT_CHANNEL)
public MessageHandler messageSender(StorageQueueOperation storageQueueOperation) {
    DefaultMessageHandler handler = new DefaultMessageHandler(STORAGE_QUEUE_NAME, storageQueueOperation);
    handler.setSendCallback(new ListenableFutureCallback<Void>() {
        @Override
        public void onSuccess(Void result) {
            LOGGER.info("Message was sent successfully.");
        }

        @Override
        public void onFailure(Throwable ex) {
            LOGGER.info("There was an error sending the message.");
        }
    });
    return handler;
}

@Autowired
StorageQueueOutboundGateway storageQueueOutboundGateway;

@MessagingGateway(defaultRequestChannel = OUTPUT_CHANNEL)
public interface StorageQueueOutboundGateway {
    void send(String text);
}

this.storageQueueOutboundGateway.send(message);

private static final String INPUT_CHANNEL = "input";

@Bean
public MessageChannel input() {
    return new DirectChannel();
}

private static final String STORAGE_QUEUE_NAME = "example";

@Bean
@InboundChannelAdapter(channel = INPUT_CHANNEL, poller = @Poller(fixedDelay = "1000"))
public StorageQueueMessageSource storageQueueMessageSource(StorageQueueOperation storageQueueOperation) {
    storageQueueOperation.setCheckpointMode(CheckpointMode.MANUAL);
    storageQueueOperation.setVisibilityTimeoutInSeconds(10);

    return new StorageQueueMessageSource(STORAGE_QUEUE_NAME, storageQueueOperation);
}

@ServiceActivator(inputChannel = INPUT_CHANNEL)
public void messageReceiver(byte[] payload, @Header(AzureHeaders.CHECKPOINTER) Checkpointer checkpointer) {
    String message = new String(payload);
    LOGGER.info("New message received: '{}'", message);
    checkpointer.success()
        .doOnError(Throwable::printStackTrace)
        .doOnSuccess(t -> LOGGER.info("Message '{}' successfully checkpointed", message))
        .subscribe();
}