csfox.com

monitoring a CCDT URL and dynamically reloading the IBM MQ connection factory in a Spring Boot application.

The core idea is to:

Externalize the CCDT URL: Manage the URL source (file, HTTP endpoint) outside the application's deployment artifact.

Monitor the Source: Implement a component that periodically checks the source for changes.

Centralize ConnectionFactory Management: Create a provider bean that holds the current ConnectionFactory instance.

Implement Reload Logic: When a change is detected, update the provider with a new ConnectionFactory instance configured with the new URL.

Restart Consumers: Critically, stop and restart JMS listeners so they pick up the new connection factory configuration and establish new connections.

Use a Proxy (Optional but Recommended): Configure beans like JmsTemplate to use a proxy ConnectionFactory that always delegates to the current instance held by the provider.

Here's a breakdown and implementation guidance:

1. Dependencies

Make sure you have the necessary IBM MQ Spring Boot starter:

<dependency>
    <groupId>com.ibm.mq</groupId>
    <artifactId>mq-jms-spring-boot-starter</artifactId>
    <version>...</version> <!-- Use the latest appropriate version -->
</dependency>

2. Configuration Properties

Define properties for the initial CCDT URL and monitoring settings.

# application.yaml or application.properties
ibm:
  mq:
    # The initial CCDT URL (e.g., file:///path/to/ccdt.json, http://config-server/ccdt)
    ccdt-url: file:///etc/mq/ccdt/AMQCLCHL.TAB
    # Optional: Queue Manager name (can often be omitted when using CCDT)
    # queue-manager: YOUR_QMGR
    # Optional: Channel name (usually defined in CCDT)
    # channel: YOUR_CHANNEL
    # ... other standard MQ properties like user, password, sslCipherSuite etc.

# Custom properties for monitoring
mq:
  ccdt:
    monitor:
      enabled: true
      # How to monitor: FILE or HTTP
      type: FILE
      # Check interval (e.g., PT1M = 1 minute)
      check-interval: PT1M
      # For HTTP type: Headers to check for changes (e.g., ETag, Last-Modified)
      # http-change-headers: ETag, Last-Modified

import jakarta.validation.constraints.NotBlank;
import jakarta.validation.constraints.NotNull;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.validation.annotation.Validated;

import java.time.Duration;

@ConfigurationProperties("mq.ccdt.monitor")
@Validated
public class CcdtMonitorProperties {

    private boolean enabled = false;

    @NotNull
    private MonitorType type = MonitorType.FILE; // Default to FILE

    @NotNull
    private Duration checkInterval = Duration.ofMinutes(1); // Default check interval

    // Optional: For HTTP monitoring
    // private List<String> httpChangeHeaders = List.of("ETag", "Last-Modified");

    public enum MonitorType { FILE, HTTP }

    // Getters and Setters
    public boolean isEnabled() { return enabled; }
    public void setEnabled(boolean enabled) { this.enabled = enabled; }
    public MonitorType getType() { return type; }
    public void setType(MonitorType type) { this.type = type; }
    public Duration getCheckInterval() { return checkInterval; }
    public void setCheckInterval(Duration checkInterval) { this.checkInterval = checkInterval; }
    // public List<String> getHttpChangeHeaders() { return httpChangeHeaders;}
    // public void setHttpChangeHeaders(List<String> httpChangeHeaders) { this.httpChangeHeaders = httpChangeHeaders; }
}

// Also need the standard IBM MQ properties class if not using the starter's defaults extensively
// For example:
// @ConfigurationProperties(prefix = "ibm.mq")
// public class IbmMqProperties {
//     private String ccdtUrl;
//     // ... other fields ...
//     // Getters & Setters
// }

3. Caching Connection Factory Provider

This bean holds the current ConnectionFactory and provides a method to reload it.

import com.ibm.mq.jms.MQConnectionFactory;
import com.ibm.msg.client.wmq.WMQConstants;
import jakarta.annotation.PostConstruct;
import jakarta.jms.ConnectionFactory;
import jakarta.jms.JMSException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.net.MalformedURLException;
import java.net.URL;
import java.util.Objects;

@Component
public class CachingMqConnectionFactoryProvider {

    private static final Logger log = LoggerFactory.getLogger(CachingMqConnectionFactoryProvider.class);

    // Inject the initial CCDT URL from properties
    @Value("${ibm.mq.ccdt-url}")
    private String initialCcdtUrl;
    // Inject other necessary properties if they aren't covered by CCDT or need explicit setting
    // @Value("${ibm.mq.queue-manager}") // Often optional with CCDT
    // private String queueManager;
    // @Value("${ibm.mq.channel}") // Often optional with CCDT
    // private String channel;
    // @Value("${ibm.mq.user}")
    // private String user; // etc.

    private volatile ConnectionFactory currentConnectionFactory;
    private volatile String currentCcdtUrl;
    private volatile URL currentCcdtUrlObject;

    @PostConstruct
    public void initialize() {
        log.info("Initializing MQ Connection Factory with CCDT URL: {}", initialCcdtUrl);
        this.currentCcdtUrl = initialCcdtUrl;
        try {
            this.currentCcdtUrlObject = new URL(initialCcdtUrl);
            this.currentConnectionFactory = createMqConnectionFactory(this.currentCcdtUrlObject);
            log.info("Initial MQ Connection Factory created successfully.");
        } catch (MalformedURLException e) {
            log.error("Invalid initial CCDT URL format: {}", initialCcdtUrl, e);
            throw new IllegalArgumentException("Invalid initial CCDT URL: " + initialCcdtUrl, e);
        } catch (JMSException e) {
            log.error("Failed to create initial MQ Connection Factory for CCDT URL: {}", initialCcdtUrl, e);
            // Decide if the application should fail to start or continue without a factory
            throw new IllegalStateException("Failed to create initial MQ Connection Factory", e);
        }
    }

    public ConnectionFactory getConnectionFactory() {
        ConnectionFactory factory = this.currentConnectionFactory;
        if (factory == null) {
            throw new IllegalStateException("MQ Connection Factory is not available. Check logs for initialization errors.");
        }
        return factory;
    }

    public String getCurrentCcdtUrl() {
        return currentCcdtUrl;
    }

    /**
     * Attempts to reload the Connection Factory with a new CCDT URL.
     *
     * @param newCcdtUrlString The new CCDT URL.
     * @return true if the factory was successfully reloaded, false otherwise (e.g., URL unchanged or error).
     */
    public synchronized boolean reload(String newCcdtUrlString) {
        if (Objects.equals(this.currentCcdtUrl, newCcdtUrlString)) {
            log.debug("CCDT URL hasn't changed: {}", newCcdtUrlString);
            return false;
        }

        log.info("Attempting to reload MQ Connection Factory. Old URL: '{}', New URL: '{}'",
                 this.currentCcdtUrl, newCcdtUrlString);

        URL newCcdtUrlObject;
        ConnectionFactory newConnectionFactory;

        try {
            newCcdtUrlObject = new URL(newCcdtUrlString);
            newConnectionFactory = createMqConnectionFactory(newCcdtUrlObject);
            log.info("Successfully created new MQ Connection Factory for URL: {}", newCcdtUrlString);
        } catch (MalformedURLException e) {
            log.error("Invalid new CCDT URL format: {}", newCcdtUrlString, e);
            return false; // Do not update if the new URL is invalid
        } catch (JMSException e) {
            log.error("Failed to create new MQ Connection Factory for CCDT URL: {}", newCcdtUrlString, e);
            // Optional: Implement retry logic here?
            return false; // Do not update if the factory creation fails
        }

        // Close the old factory *if* it's closeable and necessary.
        // MQConnectionFactory itself isn't Closeable, but connections created from it are.
        // Rely on listener container stop/start to handle connection closing.
        ConnectionFactory oldFactory = this.currentConnectionFactory;
        // Consider logging if the old factory had issues if needed

        // Atomically update
        this.currentConnectionFactory = newConnectionFactory;
        this.currentCcdtUrl = newCcdtUrlString;
        this.currentCcdtUrlObject = newCcdtUrlObject;

        log.info("MQ Connection Factory successfully reloaded with new CCDT URL: {}", newCcdtUrlString);
        return true;
    }

    private ConnectionFactory createMqConnectionFactory(URL ccdtUrl) throws JMSException {
        MQConnectionFactory factory = new MQConnectionFactory();

        // --- Core CCDT Configuration ---
        factory.setIntProperty(WMQConstants.WMQ_CONNECTION_MODE, WMQConstants.WMQ_CM_CLIENT);
        factory.setObjectProperty(WMQConstants.WMQ_CCDTURL, ccdtUrl);

        // --- Optional: Explicit QM Name (usually not needed with CCDT unless connecting to a specific QM within a group) ---
        // if (queueManager != null && !queueManager.isBlank()) {
        //    factory.setQueueManager(queueManager);
        // }

        // --- Optional: Explicit Channel (usually defined in CCDT) ---
        // if (channel != null && !channel.isBlank()) {
        //    factory.setChannel(channel);
        // }

        // --- Optional: Application Name ---
        factory.setAppName("YourAppName"); // Good for diagnostics

        // --- Optional: Security (User/Password, SSL) ---
        // if (user != null && !user.isBlank()) {
        //     factory.setStringProperty(WMQConstants.USERID, user);
        //     factory.setStringProperty(WMQConstants.PASSWORD, password); // Inject password securely
        // }
        // Configure SSL properties if needed (sslCipherSuite, etc.)
        // factory.setSSLCipherSuite("...");
        // factory.setSSLPeerName("..."); // If using peer name checking

        // --- Other Important Properties ---
        // factory.setClientReconnectOptions(WMQConstants.WMQ_CLIENT_RECONNECT); // Enable MQ client automatic reconnect
        // factory.setClientReconnectTimeout(600); // Timeout for reconnect attempts

        log.debug("Creating MQConnectionFactory with CCDT URL: {}", ccdtUrl);
        return factory;
    }
}

4. MQ Configuration (Using the Provider)

Configure your primary ConnectionFactory bean, JmsTemplate, and JmsListenerContainerFactory to use the provider. Using a proxy for the ConnectionFactory bean is recommended for components that might cache it.

import jakarta.jms.ConnectionFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.jms.DefaultJmsListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.jms.config.DefaultJmsListenerContainerFactory;
import org.springframework.jms.core.JmsTemplate;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Proxy;

@Configuration
public class MqConfiguration {

    /**
     * Creates a proxy ConnectionFactory that always delegates to the current
     * factory held by the CachingMqConnectionFactoryProvider.
     */
    @Bean
    @Primary // Make this the default ConnectionFactory for injection
    public ConnectionFactory mqConnectionFactory(CachingMqConnectionFactoryProvider provider) {
        return (ConnectionFactory) Proxy.newProxyInstance(
            ConnectionFactory.class.getClassLoader(),
            new Class<?>[]{ConnectionFactory.class}, // Add other interfaces if needed (e.g., XAConnectionFactory)
            (proxy, method, args) -> {
                ConnectionFactory target = provider.getConnectionFactory();
                try {
                    // Delegate method invocation to the *current* target factory
                    return method.invoke(target, args);
                } catch (InvocationTargetException e) {
                    // Unwrap the target exception
                    throw e.getCause();
                }
            });
    }

    @Bean
    public JmsTemplate jmsTemplate(@Qualifier("mqConnectionFactory") ConnectionFactory connectionFactory) {
        JmsTemplate template = new JmsTemplate();
        template.setConnectionFactory(connectionFactory); // Uses the proxy factory
        // Configure other JmsTemplate properties (e.g., default destination, pubSubDomain)
        return template;
    }

    @Bean(name = "jmsListenerContainerFactory")
    public DefaultJmsListenerContainerFactory jmsListenerContainerFactory(
            @Qualifier("mqConnectionFactory") ConnectionFactory connectionFactory,
            DefaultJmsListenerContainerFactoryConfigurer configurer) {
        DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();
        configurer.configure(factory, connectionFactory); // Uses the proxy factory initially
        // Configure other listener factory properties if needed
        // factory.setConcurrency("...");
        // factory.setErrorHandler(...);
        return factory;
    }
}

5. CCDT Monitor and Reloader Service

This service performs the actual monitoring and triggers the reload process.

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.jms.config.JmsListenerEndpointRegistry;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.io.InputStream;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.URI;
import java.net.URL;
import java.nio.file.*;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.time.Duration;
import java.util.Arrays;
import java.util.concurrent.ScheduledFuture;

@Service
public class CcdtMonitorService implements DisposableBean {

    private static final Logger log = LoggerFactory.getLogger(CcdtMonitorService.class);

    private final CcdtMonitorProperties monitorProperties;
    private final CachingMqConnectionFactoryProvider factoryProvider;
    private final JmsListenerEndpointRegistry listenerRegistry; // To stop/start listeners
    private final TaskScheduler taskScheduler; // For scheduled checks

    private ScheduledFuture<?> monitorTask;
    private byte[] lastFileHash = null; // For file monitoring
    private String lastETag = null;      // For HTTP monitoring
    private String lastModified = null;  // For HTTP monitoring

    public CcdtMonitorService(CcdtMonitorProperties monitorProperties,
                              CachingMqConnectionFactoryProvider factoryProvider,
                              JmsListenerEndpointRegistry listenerRegistry,
                              TaskScheduler taskScheduler) {
        this.monitorProperties = monitorProperties;
        this.factoryProvider = factoryProvider;
        this.listenerRegistry = listenerRegistry;
        this.taskScheduler = taskScheduler;
    }

    @PostConstruct
    public void startMonitoring() {
        if (!monitorProperties.isEnabled()) {
            log.info("CCDT Monitoring is disabled.");
            return;
        }

        Duration interval = monitorProperties.getCheckInterval();
        if (interval == null || interval.isNegative() || interval.isZero()) {
            log.warn("Invalid CCDT monitor check interval: {}. Monitoring disabled.", interval);
            return;
        }

        log.info("Starting CCDT URL monitoring ({}) every {}", monitorProperties.getType(), interval);
        // Schedule the check task
        this.monitorTask = taskScheduler.scheduleAtFixedRate(this::checkForCcdtChange, interval);

        // Perform an initial check immediately after startup (optional)
        // taskScheduler.schedule(this::checkForCcdtChange, Instant.now().plusSeconds(5));
    }

    public void checkForCcdtChange() {
        log.trace("Checking for CCDT change...");
        String currentUrlString = factoryProvider.getCurrentCcdtUrl();
        boolean changed;

        try {
            switch (monitorProperties.getType()) {
                case FILE:
                    changed = checkFileForChange(currentUrlString);
                    break;
                case HTTP:
                    changed = checkHttpForChange(currentUrlString);
                    break;
                default:
                    log.warn("Unsupported CCDT monitor type: {}", monitorProperties.getType());
                    changed = false; // Or handle as error
            }
        } catch (Exception e) {
            log.error("Error during CCDT check for URL '{}'", currentUrlString, e);
            changed = false; // Avoid reload on transient errors
        }

        if (changed) {
            log.info("CCDT source change detected for URL: {}", currentUrlString);
            reloadConnectionFactory(currentUrlString); // Pass the *current* URL, the provider has the logic
        } else {
            log.trace("No CCDT change detected.");
        }
    }

    private boolean checkFileForChange(String ccdtUrlString) {
        Path ccdtPath;
        try {
            // Handle potential "file:" prefix and convert to Path
             if (ccdtUrlString.startsWith("file:")) {
                ccdtPath = Paths.get(URI.create(ccdtUrlString));
            } else {
                 // Assume it's a direct path if no scheme
                 ccdtPath = Paths.get(ccdtUrlString);
             }

            if (!Files.exists(ccdtPath)) {
                log.warn("CCDT file does not exist: {}", ccdtPath);
                // If it existed before, this is a change. If not, maybe initial state.
                return lastFileHash != null; // Change if we previously had a hash
            }

            if (!Files.isRegularFile(ccdtPath)) {
                 log.warn("CCDT path is not a regular file: {}", ccdtPath);
                 return lastFileHash != null; // Change if we previously had a hash
            }

            byte[] currentHash = calculateFileHash(ccdtPath);
            if (lastFileHash == null) {
                // First check
                lastFileHash = currentHash;
                log.info("Initial CCDT file hash calculated for: {}", ccdtPath);
                return false; // No change on first check
            }

            if (!Arrays.equals(lastFileHash, currentHash)) {
                log.info("CCDT file content hash changed for: {}", ccdtPath);
                lastFileHash = currentHash;
                return true;
            } else {
                return false; // Hash matches, no change
            }

        } catch (InvalidPathException | IllegalArgumentException e) {
             log.error("Invalid CCDT file path derived from URL '{}': {}", ccdtUrlString, e.getMessage());
             return false;
        } catch (IOException | NoSuchAlgorithmException e) {
            log.error("Error accessing or hashing CCDT file '{}': {}", ccdtUrlString, e.getMessage());
            return false; // Don't trigger reload on error
        }
    }

     private byte[] calculateFileHash(Path path) throws IOException, NoSuchAlgorithmException {
        MessageDigest sha256 = MessageDigest.getInstance("SHA-256");
        try (InputStream is = Files.newInputStream(path)) {
            byte[] buffer = new byte[8192];
            int bytesRead;
            while ((bytesRead = is.read(buffer)) != -1) {
                sha256.update(buffer, 0, bytesRead);
            }
        }
        return sha256.digest();
    }


    private boolean checkHttpForChange(String ccdtUrlString) {
        // Simplified HTTP check using HEAD request and ETag/Last-Modified
        // A more robust implementation might download and hash the content if headers aren't reliable.
        HttpURLConnection connection = null;
        try {
            URL url = new URL(ccdtUrlString);
            connection = (HttpURLConnection) url.openConnection();
            connection.setRequestMethod("HEAD"); // Use HEAD to check headers without downloading body
            connection.setConnectTimeout(5000); // 5 seconds
            connection.setReadTimeout(5000);

            int responseCode = connection.getResponseCode();

            if (responseCode >= 200 && responseCode < 300) {
                String currentETag = connection.getHeaderField("ETag");
                String currentLastModified = connection.getHeaderField("Last-Modified");

                boolean etagChanged = (currentETag != null && !currentETag.equals(lastETag));
                boolean lastModifiedChanged = (currentLastModified != null && !currentLastModified.equals(lastModified));

                if (lastETag == null && lastModified == null) {
                    // First check
                    lastETag = currentETag;
                    lastModified = currentLastModified;
                    log.info("Initial CCDT HTTP headers captured for: {} (ETag: {}, Last-Modified: {})", ccdtUrlString, lastETag, lastModified);
                    return false;
                }

                if (etagChanged || lastModifiedChanged) {
                    log.info("CCDT HTTP source changed for: {} (ETag changed: {}, Last-Modified changed: {})",
                             ccdtUrlString, etagChanged, lastModifiedChanged);
                    lastETag = currentETag;
                    lastModified = currentLastModified;
                    return true;
                } else {
                    return false; // Headers match, no change detected
                }
            } else {
                log.warn("Received non-OK status code ({}) when checking CCDT URL: {}", responseCode, ccdtUrlString);
                return false; // Don't reload on server error/redirect etc.
            }

        } catch (MalformedURLException e) {
            log.error("Invalid CCDT HTTP URL format: {}", ccdtUrlString, e);
            return false;
        } catch (IOException e) {
            log.error("IOException during HTTP check for CCDT URL '{}': {}", ccdtUrlString, e.getMessage());
            return false;
        } finally {
            if (connection != null) {
                connection.disconnect();
            }
        }
    }

    private void reloadConnectionFactory(String urlUsedForCheck) {
        // It's crucial to stop listeners *before* changing the factory
        // and restart them *after* the factory is updated.
        log.info("Stopping JMS listeners before CCDT reload...");
        try {
             // Use stop(Runnable callback) for graceful shutdown if needed and supported
            listenerRegistry.stop();
            log.info("JMS listeners stopped.");
        } catch(Exception e) {
             log.error("Failed to stop JMS listeners gracefully.", e);
             // Decide if you should proceed with reload anyway or abort
             // Proceeding might leave old connections lingering until they error out
             // return; // Abort reload if stop fails?
        }


        // Attempt to reload the factory via the provider
        // The provider uses the *new* URL internally if the check passed.
        // We pass the URL used for the check just for confirmation/logging.
        boolean reloaded = factoryProvider.reload(urlUsedForCheck); // The provider has the actual logic

        // Always try to restart listeners, even if reload reported no change (e.g., if stop failed earlier)
        // or if reload failed (maybe revert to old state if possible? complex.)
        log.info("Starting JMS listeners after CCDT reload attempt (Reloaded: {})...", reloaded);
        try {
            listenerRegistry.start();
            log.info("JMS listeners started.");
        } catch (Exception e) {
             log.error("Failed to start JMS listeners after CCDT reload.", e);
             // This is problematic. Application might be partially functional.
        }
    }

    @Override
    @PreDestroy
    public void destroy() {
        if (this.monitorTask != null && !this.monitorTask.isCancelled()) {
            log.info("Stopping CCDT monitor task.");
            this.monitorTask.cancel(true); // Interrupt if running
        }
    }
}

6. Enable Scheduling

Ensure task scheduling is enabled in your Spring Boot application.

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;

@SpringBootApplication
@EnableScheduling // <--- Add this annotation
public class YourApplication {

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

}

Explanation and Key Points:

Provider Pattern: The CachingMqConnectionFactoryProvider decouples the creation and holding of the ConnectionFactory from its usage. This makes it easier to replace the underlying factory instance.

Proxy Using java.lang.reflect.Proxy ensures that components like JmsTemplate always get the latest factory instance from the provider whenever they need to create a connection or session. Without a proxy, they might cache the initial factory instance.

Listener Restart ( This is the most critical step for consumers (@JmsListener). Simply changing the ConnectionFactory bean won't affect existing listeners and their connections. You must stop the listeners (which closes their connections) and then start them again so they use the new factory configuration to establish fresh connections.

Monitoring Strategies:

FILE: Checks the file's content hash (SHA-256). This detects changes even if the timestamp doesn't update reliably. Assumes the ccdt-url property points to the file path (handles file: scheme).

HTTP: Uses a HEAD request to check ETag or Last-Modified headers. This is efficient but relies on the server providing these headers correctly and consistently. If not, you might need to GET the content and hash it, similar to the file approach.

Error Handling: The monitor service includes basic error handling to prevent unnecessary reloads due to temporary network issues or file access problems. The provider handles errors during factory creation. Consider adding more sophisticated retry logic or alerting if failures persist.

Thread Safety: The CachingMqConnectionFactoryProvider uses volatile and synchronized on the reload method to ensure thread-safe updates to the factory instance and URL. The monitor runs in a separate scheduled thread.

Configuration: All key parameters (initial URL, monitoring type, interval) are externalized to application.properties/yaml.

Graceful Shutdown: The DisposableBean interface and @PreDestroy ensure the monitoring task is cancelled when the application shuts down.

This setup provides a dynamic and robust way to handle changes to your IBM MQ CCDT URL without requiring application restarts. Remember to test thoroughly, especially the listener restart logic and error handling scenarios.

csfox.com

Thursday, April 10, 2025

using helm and kubectl to deploy ccdt

Dynamic IBM MQ CCDT URL connection