How I Built a System to Transfer Messages Across Cloud Providers

The Challenge: Cross-Cloud Messaging

During a recent cloud transformation initiative, I faced a significant challenge: enabling reliable and efficient communication between different cloud providers’ messaging systems. These systems—each with their own protocols and architecture—often lack native compatibility, making direct integration difficult or impossible. The goal was to create a solution that could bridge this gap, ensuring business continuity while enabling cross-cloud workflows.

The Solution: EventBridge with Apache Camel

To address this challenge, I developed an in-house system called EventBridge, powered by Apache Camel. Hosted on a Kubernetes cluster, EventBridge allows customers to easily configure route bridges between messaging services, while Apache Camel handles the underlying complexities of integration.

This approach allowed us to connect on-premises systems, such as Kafka, with cloud services like Azure Service Bus, AWS SQS, and GCP Pub/Sub. Apache Camel’s rich set of connectors ensured that the underlying protocols and configurations for each service were abstracted away, enabling seamless communication.

Configuration-Driven Routing for Customers

With EventBridge, customers could define and manage route bridges through configuration. Apache Camel handled all the underlying components, such as connection details, protocols, and data transformation. Here’s an example of how a configuring looks like:

Example: Congfiguration

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
package config;

import io.smallrye.config.ConfigMapping;
import io.smallrye.config.WithDefault;

import java.util.List;
import java.util.Optional;

@ConfigMapping(prefix = "config")
public interface RouteConfig {

    public List<CamelRouteConfiguration> CAMEL_ROUTE_CONFIGURATION_LIST();

    interface CamelRouteConfiguration {
        String routeName();

        @WithDefault("false")
        Boolean transformMessage();

        @WithDefault("true")
        Boolean enable ();

        Metadata source();

        Metadata target();
    }

    interface Metadata {
        Optional<String> topic ();
        CloudServiceQueue service();
        String topicUniqueIdentifier();
    }

    enum CloudServiceQueue {
        AZURE_SERVICEBUS, ONPREM;
    }
}

Example: Camel Template for Azure Service Bus

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
package template;

import org.apache.camel.builder.RouteBuilder;

/**
 * Azure Service Bus
 * Protocols Supported: AMQP
 */
public class AzureServiceBusTemplate extends RouteBuilder implements RouteTemplate {

    private String PRODUCER_ROUTE_TEMPLATE_NAME = "AZURE_SERVICE_BUS_PRODUCER";

    private String PRODUCER_ROUTE_ENTRY_POINT = "direct:AZURE_SERVICE_BUS_PRODUCER";

    private String CONSUMER_ROUTE_TEMPLATE_NAME = "AZURE_SERVICE_BUS_CONSUMER";

    @Override
    public void configure() throws Exception {
        // Empty on purpose
    }

    @Override
    public String getProducerTemplateName() {
        return PRODUCER_ROUTE_TEMPLATE_NAME;
    }

    @Override
    public String getConsumerTemplateName() {
        return CONSUMER_ROUTE_TEMPLATE_NAME;
    }

    @Override
    public String producerRouteEntryPointName() {
        return PRODUCER_ROUTE_ENTRY_POINT;
    }

    @Override
    public void buildProducerRoute() {
        routeTemplate(this.PRODUCER_ROUTE_TEMPLATE_NAME)
                .templateParameter("topic", "TOPIC-NOT-DEFINED")

                // Definition
                .from(this.PRODUCER_ROUTE_ENTRY_POINT)
                .to("azure-servicebus:{{topic}}")
                .log("Topic: {{topic}}")
                .log("${body}");
    }

    @Override
    public void buildConsumerRoute() {
        routeTemplate(this.CONSUMER_ROUTE_TEMPLATE_NAME)
                .templateParameter("poll-topic", "POLL-TOPIC-NOT-DEFINED")
                .templateParameter("destination", "DESTINATION-NOT-DEFINED")

                // Definition
                .from("azure-servicebus:{{poll-topic}}")
                .to("{{destination}}")
                .log("Destination: {{destination}}")
                .log("${body}");

    }
}

Example: Camel Template for On premp Kafka Cluster

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
package template;

import org.apache.camel.builder.RouteBuilder;

/**
 * On prem Kafka Cluster
 * Supported Protocol: Kafka
 */
public class OnPremKafkaTemplate extends RouteBuilder implements RouteTemplate {
    private String PRODUCER_ROUTE_TEMPLATE_NAME = "KAFKA_PRODUCER";

    private String PRODUCER_ROUTE_ENTRY_POINT = "direct:KAFKA_PRODUCER";

    private String CONSUMER_ROUTE_TEMPLATE_NAME = "KAFKA_CONSUMER";


    @Override
    public void configure() throws Exception {
        // Empty on purpose
    }

    @Override
    public String getProducerTemplateName() {
        return PRODUCER_ROUTE_TEMPLATE_NAME;
    }

    @Override
    public String getConsumerTemplateName() {
        return CONSUMER_ROUTE_TEMPLATE_NAME;
    }

    @Override
    public String producerRouteEntryPointName(){
        return PRODUCER_ROUTE_ENTRY_POINT;
    }

    @Override
    public void buildProducerRoute() {
        routeTemplate(this.PRODUCER_ROUTE_TEMPLATE_NAME)
                .templateParameter("topic", "TOPIC-NOT-DEFINED")

                // Definition
                .from(this.PRODUCER_ROUTE_ENTRY_POINT)
                .to("kafka:{{topic}}")
                .log("Topic: {{topic}}")
                .log("${body}");
    }

    @Override
    public void buildConsumerRoute() {
        routeTemplate(this.CONSUMER_ROUTE_TEMPLATE_NAME)
                .templateParameter("poll-topic", "POLL-TOPIC-NOT-DEFINED")
                .templateParameter("destination", "DESTINATION-NOT-DEFINED")

                // Definition
                .from("kafka:{{poll-topic}}")
                .to("{{destination}}")
                .log("Destination: {{destination}}")
                .log("${body}");
    }
}

How It Works

The RouteConfig interface allows defining dynamic routes for Apache Camel using configuration files (e.g., YAML) or environment variables. Each route can specify its source, target, and additional metadata, providing flexibility and modularity in setting up bridges.

Key Features

Dynamic Routing:
- Enables defining multiple routes in a centralized configuration file under the config prefix.
- Each route includes details for:
  - Source: The originating system (e.g., Kafka).
  - Target: The destination system (e.g., Azure Service Bus).
Metadata for Flexibility:
- Each route has detailed metadata, including:
  - Topic Name (Optional): The topic to route messages through.
  - Cloud Service Queue: Specifies the service type (e.g., AZURE_SERVICEBUS, ONPREM).
  - Unique Identifier: A string to uniquely identify each topic.
Conditional Routing:
- Routes can be enabled or disabled using the enable flag, making it easy to toggle routes during testing or deployment.
- Supports message transformation through the transformMessage flag, which allows modifying the message payload if needed. For example, you can dynamically specify a transformer class by including it in the classpath.

Operational benefits

Centralized Configurations:
- All route definitions are stored in a single configuration file, reducing complexity and improving manageability.
Dynamic Deployment:
- As new routes are added or updated, they are picked up without modifying the core logic of the system.
Seamless Integration:
- This configuration feeds into Camel templates to dynamically create bridges between systems such as:
  - Kafka ↔ Azure Service Bus
  - On-Prem Kafka ↔ Cloud Services

By centralizing and standardizing route configurations, RouteConfig ensures scalability and ease of management in multi-cloud and hybrid-cloud messaging systems.

Addressing Infinite Loops

One of the major issues we encountered during implementation was the risk of infinite loops. This could occur when a bridged message was picked up again by the system, causing it to loop through the routes repeatedly.

To prevent this, we introduced a custom metadata tagging for messages being bridged, using a tag called x-eventbridge-sync. This metadata allowed us to:

Tag messages that had already been bridged:
We included metadata with a unique identifier to track messages as they moved through the routes.
Filter out bridged messages:
Routes were configured to ignore messages with the x-eventbridge-sync tag, ensuring they wouldn’t loop back into the system.
Handle unprocessed messages gracefully:
In cases where messages were misrouted or encountered errors, they were sent to an empty queue for further inspection or simply logged for debugging.

Conclusion

EventBridge, powered by Apache Camel, made it easier to handle cross-cloud messaging by simplifying the configuration of routes and abstracting the complexities of integration. This solution was lightweight, flexible, and easy to extend, making it a practical approach for connecting on-prem and cloud services without significant overhead.