Data Management and Streaming Patterns in Microservices: A Comprehensive Guide#

1
{
2
  "type": "record",
3
  "name": "OrderEvent",
4
  "namespace": "com.company.orders.events",
5
  "doc": "Schema for order events in the streaming platform",
6
  "fields": [
7
    {
8
      "name": "eventId",
9
      "type": "string",
10
      "doc": "Unique event identifier"
11
    },
12
    {
13
      "name": "eventType",
14
      "type": {
15
        "type": "enum",
16
        "name": "OrderEventType",
17
        "symbols": ["CREATED", "UPDATED", "CANCELLED", "SHIPPED", "DELIVERED"]
18
      }
19
    },
20
    {
21
      "name": "orderId",
22
      "type": "string",
23
      "doc": "Order identifier"
24
    },
25
    {
26
      "name": "customerId",
27
      "type": "string",
28
      "doc": "Customer identifier"
29
    },
30
    {
31
      "name": "orderData",
32
      "type": {
33
        "type": "record",
34
        "name": "OrderData",
35
        "fields": [
36
          { "name": "totalAmount", "type": "double" },
37
          { "name": "currency", "type": "string" },
38
          { "name": "items", "type": { "type": "array", "items": "string" } },
39
          { "name": "shippingAddress", "type": "string" }
40
        ]
41
      }
42
    },
43
    {
44
      "name": "timestamp",
45
      "type": "long",
46
      "logicalType": "timestamp-millis",
47
      "doc": "Event timestamp in milliseconds"
48
    },
49
    {
50
      "name": "metadata",
51
      "type": {
52
        "type": "map",
53
        "values": "string"
54
      },
55
      "default": {}
56
    }
57
  ]
58
}

1
{
2
  "name": "user-database-connector",
3
  "config": {
4
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
5
    "database.hostname": "postgres-primary",
6
    "database.port": "5432",
7
    "database.user": "debezium",
8
    "database.password": "${file:/opt/kafka/secrets/db-password.txt:password}",
9
    "database.dbname": "userdb",
10
    "database.server.name": "user-db",
11
    "table.include.list": "public.users,public.user_profiles,public.user_preferences",
12
    "plugin.name": "pgoutput",
13
    "slot.name": "debezium_user_slot",
14
    "publication.name": "debezium_publication",
15
    "transforms": "route,unwrap",
16
    "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
17
    "transforms.route.regex": "([^.]+)\\.([^.]+)\\.([^.]+)",
18
    "transforms.route.replacement": "$3",
19
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
20
    "transforms.unwrap.drop.tombstones": "false",
21
    "transforms.unwrap.delete.handling.mode": "rewrite",
22
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
23
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
24
    "key.converter.schemas.enable": "false",
25
    "value.converter.schemas.enable": "false",
26
    "snapshot.mode": "initial",
27
    "decimal.handling.mode": "double",
28
    "include.schema.changes": "true",
29
    "provide.transaction.metadata": "true",
30
    "max.batch.size": "2048",
31
    "max.queue.size": "81920"
32
  }
33
}

1
@Component
2
@Slf4j
3
public class UserCDCEventProcessor {
4

5
    @KafkaListener(topics = "users", groupId = "user-cdc-processor")
6
    public void processUserChange(
7
            @Payload UserChangeEvent event,
8
            @Header Map<String, Object> headers) {
9

10
        log.info("Processing user change event: {}", event.getEventType());
11

12
        switch (event.getEventType()) {
13
            case CREATE:
14
                handleUserCreation(event);
15
                break;
16
            case UPDATE:
17
                handleUserUpdate(event);
18
                break;
19
            case DELETE:
20
                handleUserDeletion(event);
21
                break;
22
        }
23
    }
24

25
    private void handleUserCreation(UserChangeEvent event) {
26
        // Update search index
27
        searchService.indexUser(event.getAfter());
28

29
        // Invalidate cache
30
        cacheService.evictUserCache(event.getAfter().getId());
31

32
        // Update materialized views
33
        materializedViewService.updateUserView(event.getAfter());
34

35
        // Trigger downstream events
36
        eventPublisher.publishEvent(new UserCreatedEvent(event.getAfter()));
37
    }
38

39
    private void handleUserUpdate(UserChangeEvent event) {
40
        UserData before = event.getBefore();
41
        UserData after = event.getAfter();
42

43
        // Compare changes and update accordingly
44
        if (!before.getEmail().equals(after.getEmail())) {
45
            emailChangeService.handleEmailChange(before.getId(),
46
                before.getEmail(), after.getEmail());
47
        }
48

49
        if (!before.getPreferences().equals(after.getPreferences())) {
50
            recommendationService.updateUserPreferences(after.getId(),
51
                after.getPreferences());
52
        }
53

54
        // Update search index with changes only
55
        searchService.updateUser(after, getChangedFields(before, after));
56
    }
57

58
    private void handleUserDeletion(UserChangeEvent event) {
59
        String userId = event.getBefore().getId();
60

61
        // Remove from search index
62
        searchService.removeUser(userId);
63

64
        // Clear all related caches
65
        cacheService.evictAllUserCaches(userId);
66

67
        // Clean up materialized views
68
        materializedViewService.removeUserData(userId);
69

70
        // Publish deletion event
71
        eventPublisher.publishEvent(new UserDeletedEvent(userId));
72
    }
73
}

1
@Entity
2
@Table(name = "event_store")
3
public class EventStore {
4
    @Id
5
    private String eventId;
6
    private String aggregateId;
7
    private String eventType;
8
    private String eventData;
9
    private Long version;
10
    private Instant timestamp;
11
    private String metadata;
12

13
    // Getters, setters, constructors
14
}
15

16
@Service
17
@Transactional
18
public class EventSourcingService {
19

20
    @Autowired
21
    private EventStoreRepository eventStoreRepository;
22

23
    @Autowired
24
    private KafkaTemplate<String, Object> kafkaTemplate;
25

26
    public void saveEvent(DomainEvent event) {
27
        // Save to event store
28
        EventStore eventStore = new EventStore();
29
        eventStore.setEventId(UUID.randomUUID().toString());
30
        eventStore.setAggregateId(event.getAggregateId());
31
        eventStore.setEventType(event.getClass().getSimpleName());
32
        eventStore.setEventData(jsonUtils.toJson(event));
33
        eventStore.setVersion(getNextVersion(event.getAggregateId()));
34
        eventStore.setTimestamp(Instant.now());
35

36
        eventStoreRepository.save(eventStore);
37

38
        // Publish to event stream
39
        kafkaTemplate.send("domain-events", event.getAggregateId(), event);
40
    }
41

42
    public List<DomainEvent> getEvents(String aggregateId) {
43
        return eventStoreRepository.findByAggregateIdOrderByVersion(aggregateId)
44
                .stream()
45
                .map(this::deserializeEvent)
46
                .collect(Collectors.toList());
47
    }
48

49
    public <T> T reconstructAggregate(String aggregateId, Class<T> aggregateClass) {
50
        List<DomainEvent> events = getEvents(aggregateId);
51
        T aggregate = createEmptyAggregate(aggregateClass);
52

53
        events.forEach(event -> applyEvent(aggregate, event));
54

55
        return aggregate;
56
    }
57
}

1
@Component
2
public class OrderProcessingSaga {
3

4
    @SagaOrchestrationStart
5
    @KafkaListener(topics = "order-created")
6
    public void startOrderProcessing(OrderCreatedEvent event) {
7
        SagaTransaction saga = SagaTransaction.builder()
8
                .sagaType("ORDER_PROCESSING")
9
                .correlationId(event.getOrderId())
10
                .build();
11

12
        // Step 1: Reserve inventory
13
        sagaManager.executeStep(saga, "RESERVE_INVENTORY",
14
            new ReserveInventoryCommand(event.getOrderId(), event.getItems()));
15
    }
16

17
    @SagaOrchestrationContinue
18
    @KafkaListener(topics = "inventory-reserved")
19
    public void handleInventoryReserved(InventoryReservedEvent event) {
20
        // Step 2: Process payment
21
        sagaManager.executeStep(getSaga(event.getOrderId()), "PROCESS_PAYMENT",
22
            new ProcessPaymentCommand(event.getOrderId(), event.getAmount()));
23
    }
24

25
    @SagaOrchestrationContinue
26
    @KafkaListener(topics = "payment-processed")
27
    public void handlePaymentProcessed(PaymentProcessedEvent event) {
28
        // Step 3: Update order status
29
        sagaManager.executeStep(getSaga(event.getOrderId()), "CONFIRM_ORDER",
30
            new ConfirmOrderCommand(event.getOrderId()));
31
    }
32

33
    @SagaOrchestrationContinue
34
    @KafkaListener(topics = "order-confirmed")
35
    public void handleOrderConfirmed(OrderConfirmedEvent event) {
36
        // Saga completed successfully
37
        sagaManager.completeSaga(getSaga(event.getOrderId()));
38
    }
39

40
    // Compensation handlers
41
    @SagaOrchestrationCompensate
42
    @KafkaListener(topics = "payment-failed")
43
    public void compensateInventoryReservation(PaymentFailedEvent event) {
44
        sagaManager.compensateStep(getSaga(event.getOrderId()), "RESERVE_INVENTORY",
45
            new ReleaseInventoryCommand(event.getOrderId()));
46
    }
47
}

1
public abstract class DomainEvent {
2
    private final String eventId;
3
    private final String aggregateId;
4
    private final Instant occurredOn;
5
    private final Long version;
6

7
    protected DomainEvent(String aggregateId, Long version) {
8
        this.eventId = UUID.randomUUID().toString();
9
        this.aggregateId = aggregateId;
10
        this.occurredOn = Instant.now();
11
        this.version = version;
12
    }
13

14
    // Getters
15
    public String getEventId() { return eventId; }
16
    public String getAggregateId() { return aggregateId; }
17
    public Instant getOccurredOn() { return occurredOn; }
18
    public Long getVersion() { return version; }
19
}
20

21
@JsonTypeName("UserRegistered")
22
public class UserRegisteredEvent extends DomainEvent {
23
    private final String userId;
24
    private final String email;
25
    private final String firstName;
26
    private final String lastName;
27
    private final Instant registrationTime;
28

29
    public UserRegisteredEvent(String userId, String email,
30
                              String firstName, String lastName) {
31
        super(userId, 1L);
32
        this.userId = userId;
33
        this.email = email;
34
        this.firstName = firstName;
35
        this.lastName = lastName;
36
        this.registrationTime = Instant.now();
37
    }
38

39
    // Getters
40
}
41

42
@Component
43
public class EventPublisher {
44

45
    @Autowired
46
    private KafkaTemplate<String, DomainEvent> kafkaTemplate;
47

48
    @Autowired
49
    private EventStoreService eventStoreService;
50

51
    @EventListener
52
    @Async
53
    public void handleDomainEvent(DomainEvent event) {
54
        try {
55
            // Store event
56
            eventStoreService.saveEvent(event);
57

58
            // Publish to stream
59
            String topicName = getTopicName(event);
60
            kafkaTemplate.send(topicName, event.getAggregateId(), event)
61
                    .addCallback(
62
                            result -> log.info("Event published successfully: {}",
63
                                event.getEventId()),
64
                            failure -> log.error("Failed to publish event: {}",
65
                                event.getEventId(), failure)
66
                    );
67
        } catch (Exception e) {
68
            log.error("Error handling domain event: {}", event.getEventId(), e);
69
            // Implement retry logic or dead letter queue
70
        }
71
    }
72

73
    private String getTopicName(DomainEvent event) {
74
        return event.getClass().getSimpleName()
75
                .replaceAll("([a-z])([A-Z])", "$1-$2")
76
                .toLowerCase();
77
    }
78
}

1
@Component
2
public class OrderAnalyticsJob {
3

4
    public void executeJob() throws Exception {
5
        StreamExecutionEnvironment env = StreamExecutionEnvironment
6
                .getExecutionEnvironment();
7

8
        // Configure checkpointing
9
        env.enableCheckpointing(30000);
10
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
11
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(5000);
12
        env.getCheckpointConfig().setCheckpointTimeout(60000);
13

14
        // Configure Kafka source
15
        Properties kafkaProps = new Properties();
16
        kafkaProps.setProperty("bootstrap.servers", "kafka:9092");
17
        kafkaProps.setProperty("group.id", "order-analytics");
18

19
        FlinkKafkaConsumer<OrderEvent> orderSource = new FlinkKafkaConsumer<>(
20
                "order-events",
21
                new OrderEventDeserializationSchema(),
22
                kafkaProps
23
        );
24

25
        DataStream<OrderEvent> orderStream = env.addSource(orderSource);
26

27
        // Real-time order analytics
28
        DataStream<OrderMetrics> orderMetrics = orderStream
29
                .filter(event -> event.getEventType() == OrderEventType.CREATED)
30
                .keyBy(OrderEvent::getCustomerId)
31
                .window(TumblingEventTimeWindows.of(Time.minutes(5)))
32
                .aggregate(new OrderAggregateFunction());
33

34
        // Fraud detection
35
        DataStream<FraudAlert> fraudAlerts = orderStream
36
                .keyBy(OrderEvent::getCustomerId)
37
                .process(new FraudDetectionProcessFunction());
38

39
        // Real-time recommendations
40
        DataStream<ProductRecommendation> recommendations = orderStream
41
                .connect(userPreferencesStream)
42
                .keyBy(OrderEvent::getCustomerId, UserPreference::getUserId)
43
                .process(new RecommendationProcessFunction());
44

45
        // Output to sinks
46
        orderMetrics.addSink(new ElasticsearchSink<>(getElasticsearchConfig()));
47
        fraudAlerts.addSink(new FlinkKafkaProducer<>("fraud-alerts",
48
                new FraudAlertSerializationSchema(), kafkaProps));
49
        recommendations.addSink(new RedisSink<>(getRedisConfig()));
50

51
        env.execute("Order Analytics Job");
52
    }
53
}
54

55
public class OrderAggregateFunction
56
        implements AggregateFunction<OrderEvent, OrderAccumulator, OrderMetrics> {
57

58
    @Override
59
    public OrderAccumulator createAccumulator() {
60
        return new OrderAccumulator();
61
    }
62

63
    @Override
64
    public OrderAccumulator add(OrderEvent event, OrderAccumulator accumulator) {
65
        accumulator.addOrder(event);
66
        return accumulator;
67
    }
68

69
    @Override
70
    public OrderMetrics getResult(OrderAccumulator accumulator) {
71
        return OrderMetrics.builder()
72
                .customerId(accumulator.getCustomerId())
73
                .orderCount(accumulator.getOrderCount())
74
                .totalAmount(accumulator.getTotalAmount())
75
                .averageOrderValue(accumulator.getAverageOrderValue())
76
                .windowStart(accumulator.getWindowStart())
77
                .windowEnd(accumulator.getWindowEnd())
78
                .build();
79
    }
80

81
    @Override
82
    public OrderAccumulator merge(OrderAccumulator a, OrderAccumulator b) {
83
        return a.merge(b);
84
    }
85
}

1
public class FraudDetectionProcessFunction
2
        extends KeyedProcessFunction<String, OrderEvent, FraudAlert> {
3

4
    private static final double FRAUD_THRESHOLD = 1000.0;
5
    private static final long TIME_WINDOW = 60000; // 1 minute
6

7
    private transient ValueState<Double> totalAmountState;
8
    private transient ValueState<Integer> orderCountState;
9
    private transient ValueState<Long> windowStartState;
10

11
    @Override
12
    public void open(Configuration parameters) {
13
        ValueStateDescriptor<Double> amountDescriptor =
14
                new ValueStateDescriptor<>("totalAmount", Double.class);
15
        totalAmountState = getRuntimeContext().getState(amountDescriptor);
16

17
        ValueStateDescriptor<Integer> countDescriptor =
18
                new ValueStateDescriptor<>("orderCount", Integer.class);
19
        orderCountState = getRuntimeContext().getState(countDescriptor);
20

21
        ValueStateDescriptor<Long> windowDescriptor =
22
                new ValueStateDescriptor<>("windowStart", Long.class);
23
        windowStartState = getRuntimeContext().getState(windowDescriptor);
24
    }
25

26
    @Override
27
    public void processElement(OrderEvent event, Context ctx,
28
                              Collector<FraudAlert> out) throws Exception {
29

30
        long currentTime = ctx.timestamp();
31
        Long windowStart = windowStartState.value();
32

33
        // Initialize or reset window
34
        if (windowStart == null || currentTime - windowStart > TIME_WINDOW) {
35
            windowStartState.update(currentTime);
36
            totalAmountState.update(0.0);
37
            orderCountState.update(0);
38

39
            // Set timer for window cleanup
40
            ctx.timerService().registerEventTimeTimer(currentTime + TIME_WINDOW);
41
        }
42

43
        // Update state
44
        Double currentTotal = totalAmountState.value();
45
        Integer currentCount = orderCountState.value();
46

47
        totalAmountState.update(currentTotal + event.getAmount());
48
        orderCountState.update(currentCount + 1);
49

50
        // Check for fraud patterns
51
        if (totalAmountState.value() > FRAUD_THRESHOLD) {
52
            FraudAlert alert = FraudAlert.builder()
53
                    .customerId(event.getCustomerId())
54
                    .alertType(FraudAlertType.HIGH_VALUE_TRANSACTIONS)
55
                    .totalAmount(totalAmountState.value())
56
                    .orderCount(orderCountState.value())
57
                    .timeWindow(TIME_WINDOW)
58
                    .timestamp(currentTime)
59
                    .build();
60

61
            out.collect(alert);
62
        }
63

64
        // Check for rapid fire orders
65
        if (orderCountState.value() > 10) {
66
            FraudAlert alert = FraudAlert.builder()
67
                    .customerId(event.getCustomerId())
68
                    .alertType(FraudAlertType.RAPID_FIRE_ORDERS)
69
                    .orderCount(orderCountState.value())
70
                    .timeWindow(TIME_WINDOW)
71
                    .timestamp(currentTime)
72
                    .build();
73

74
            out.collect(alert);
75
        }
76
    }
77

78
    @Override
79
    public void onTimer(long timestamp, OnTimerContext ctx,
80
                       Collector<FraudAlert> out) {
81
        // Clean up expired state
82
        totalAmountState.clear();
83
        orderCountState.clear();
84
        windowStartState.clear();
85
    }
86
}

1
import org.apache.spark.sql.SparkSession
2
import org.apache.spark.sql.streaming.Trigger
3
import io.delta.tables._
4

5
object MicroservicesDataLake {
6

7
  def main(args: Array[String]): Unit = {
8
    val spark = SparkSession.builder()
9
      .appName("Microservices Data Lake")
10
      .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")
11
      .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")
12
      .getOrCreate()
13

14
    import spark.implicits._
15

16
    // Stream from Kafka to Delta Lake
17
    val orderStream = spark
18
      .readStream
19
      .format("kafka")
20
      .option("kafka.bootstrap.servers", "kafka:9092")
21
      .option("subscribe", "order-events")
22
      .load()
23
      .selectExpr("CAST(value AS STRING)")
24
      .select(from_json($"value", orderEventSchema).as("data"))
25
      .select("data.*")
26

27
    // Write to Delta Lake with schema evolution
28
    val deltaWriter = orderStream
29
      .writeStream
30
      .format("delta")
31
      .outputMode("append")
32
      .option("checkpointLocation", "/delta/checkpoints/orders")
33
      .option("mergeSchema", "true")
34
      .trigger(Trigger.ProcessingTime("30 seconds"))
35
      .start("/delta/tables/orders")
36

37
    // Create materialized views
38
    createOrderAnalyticsView(spark)
39

40
    // Real-time aggregations
41
    val orderMetrics = orderStream
42
      .groupBy(
43
        window($"timestamp", "5 minutes"),
44
        $"customerId"
45
      )
46
      .agg(
47
        count("*").as("orderCount"),
48
        sum("totalAmount").as("totalRevenue"),
49
        avg("totalAmount").as("avgOrderValue")
50
      )
51

52
    orderMetrics
53
      .writeStream
54
      .format("delta")
55
      .outputMode("update")
56
      .option("checkpointLocation", "/delta/checkpoints/metrics")
57
      .start("/delta/tables/order_metrics")
58

59
    deltaWriter.awaitTermination()
60
  }
61

62
  def createOrderAnalyticsView(spark: SparkSession): Unit = {
63
    val orders = DeltaTable.forPath(spark, "/delta/tables/orders")
64
    val customers = DeltaTable.forPath(spark, "/delta/tables/customers")
65

66
    // Merge customer data with orders
67
    orders.alias("orders")
68
      .merge(
69
        customers.toDF.alias("customers"),
70
        "orders.customerId = customers.id"
71
      )
72
      .whenMatched
73
      .updateExpr(Map(
74
        "customerSegment" -> "customers.segment",
75
        "customerLifetimeValue" -> "customers.lifetimeValue"
76
      ))
77
      .execute()
78

79
    // Create aggregated view
80
    spark.sql("""
81
      CREATE OR REPLACE TEMPORARY VIEW order_analytics AS
82
      SELECT
83
        customerId,
84
        customerSegment,
85
        DATE(timestamp) as orderDate,
86
        COUNT(*) as dailyOrders,
87
        SUM(totalAmount) as dailyRevenue,
88
        AVG(totalAmount) as avgOrderValue,
89
        MAX(totalAmount) as maxOrderValue
90
      FROM delta.`/delta/tables/orders`
91
      WHERE timestamp >= current_date() - INTERVAL 30 DAYS
92
      GROUP BY customerId, customerSegment, DATE(timestamp)
93
    """)
94
  }
95
}