如何将Kafka的三个KStream（BEGIN、CONTINUE、END主题）进行连接？

Hey there! Connecting three Kafka topics ("BEGIN", "CONTINUE", "END") into a single combined message is totally feasible—let’s break this down with practical ideas and code examples, building on the two-topic joins you already know.

First, a critical prerequisite: we’ll assume each message across the three topics shares a common correlation key (like requestId). This is how we’ll match which BEGIN, CONTINUE, and END messages belong to the same logical transaction or request—without this, we can’t reliably group messages together.

You can extend two-topic join logic by splitting the three-topic problem into two sequential two-topic joins, or use windowed aggregation to collect all three message types for a key before emitting the combined result. Let’s dive into both methods.

Method 1: Sequential Left Joins (For Ordered Message Arrival) #

If your messages typically arrive in BEGIN → CONTINUE → END order, sequential joins work great. We’ll first combine BEGIN and CONTINUE into an intermediate result, then join that with END to get the final combined model.

Example with Kafka Streams (Java) #

First, define your data models (adjust for your preferred language like Scala/Python):

// Raw event models
class BeginEvent {
    String requestId;
    String startTime;
    String data;
    // Getters, setters, and constructors
}

class ContinueEvent {
    String requestId;
    String processTime;
    String data;
    // Getters, setters, and constructors
}

class EndEvent {
    String requestId;
    String endTime;
    String data;
    // Getters, setters, and constructors
}

// Intermediate model for BEGIN + CONTINUE
class CombinedPartial {
    String requestId;
    String startTime;
    String processTime;
    List<String> dataParts;
    // Getters, setters, and constructors
}

// Final combined result model
class CombinedEvent {
    String requestId;
    String startTime;
    String processTime;
    String endTime;
    List<String> dataParts;
    // Getters, setters, and constructors
}

Next, build the Streams topology:

StreamsBuilder builder = new StreamsBuilder();

// Read each topic into a stream, keyed by requestId
KStream<String, BeginEvent> beginStream = builder.stream("BEGIN", Consumed.with(Serdes.String(), new JsonSerde<>(BeginEvent.class)))
        .selectKey((ignoredKey, event) -> event.getRequestId());

KStream<String, ContinueEvent> continueStream = builder.stream("CONTINUE", Consumed.with(Serdes.String(), new JsonSerde<>(ContinueEvent.class)))
        .selectKey((ignoredKey, event) -> event.getRequestId());

KStream<String, EndEvent> endStream = builder.stream("END", Consumed.with(Serdes.String(), new JsonSerde<>(EndEvent.class)))
        .selectKey((ignoredKey, event) -> event.getRequestId());

// Step 1: Join BEGIN and CONTINUE into partial results
KStream<String, CombinedPartial> partialStream = beginStream.leftJoin(
        continueStream,
        (begin, cont) -> new CombinedPartial(
                begin.getRequestId(),
                begin.getStartTime(),
                cont.getProcessTime(),
                Arrays.asList(begin.getData(), cont.getData())
        ),
        JoinWindows.of(Duration.ofMinutes(5)), // Window to catch delayed CONTINUE messages
        Joined.with(Serdes.String(), new JsonSerde<>(BeginEvent.class), new JsonSerde<>(ContinueEvent.class))
);

// Step 2: Join partial results with END to get final combined events
KStream<String, CombinedEvent> finalStream = partialStream.leftJoin(
        endStream,
        (partial, end) -> {
            List<String> allData = new ArrayList<>(partial.getDataParts());
            allData.add(end.getData());
            return new CombinedEvent(
                    partial.getRequestId(),
                    partial.getStartTime(),
                    partial.getProcessTime(),
                    end.getEndTime(),
                    allData
            );
        },
        JoinWindows.of(Duration.ofMinutes(5)),
        Joined.with(Serdes.String(), new JsonSerde<>(CombinedPartial.class), new JsonSerde<>(EndEvent.class))
);

// Output the final combined events to a new topic
finalStream.to("COMBINED_RESULT", Produced.with(Serdes.String(), new JsonSerde<>(CombinedEvent.class)));

// Start the Streams application
KafkaStreams streams = new KafkaStreams(builder.build(), yourStreamsConfig);
streams.start();

Method 2: Windowed Aggregation (For Unordered Message Arrival) #

If messages might arrive out of order (e.g., END comes before CONTINUE), use windowed aggregation to collect all three event types for a key within a time window. Once all three are present, emit the combined result.

Example with Kafka Streams (Java) #

First, define a generic event wrapper to track message origin:

class GenericEvent {
    String type; // "BEGIN", "CONTINUE", "END"
    String payload; // Raw JSON string of the original message
    // Getters, setters, and constructors
}

Then build the aggregation topology:

StreamsBuilder builder = new StreamsBuilder();
ObjectMapper objectMapper = new ObjectMapper();

// Combine all three topics into a single stream, keyed by requestId and tagged with event type
KStream<String, GenericEvent> unifiedStream = builder.stream(Arrays.asList("BEGIN", "CONTINUE", "END"), Consumed.with(Serdes.String(), Serdes.String()))
        .map((ignoredKey, rawPayload) -> {
            JsonNode node = objectMapper.readTree(rawPayload);
            String requestId = node.get("requestId").asText();
            String eventType = node.has("startTime") ? "BEGIN" : node.has("processTime") ? "CONTINUE" : "END";
            return new KeyValue<>(requestId, new GenericEvent(eventType, rawPayload));
        });

// Aggregate events per requestId within a 5-minute window
KTable<Windowed<String>, CombinedEvent> aggregatedTable = unifiedStream.groupByKey()
        .windowedBy(TimeWindows.of(Duration.ofMinutes(5)))
        .aggregate(
                CombinedEvent::new, // Initialize empty combined event
                (requestId, genericEvent, combinedEvent) -> {
                    JsonNode node = objectMapper.readTree(genericEvent.getPayload());
                    combinedEvent.setRequestId(requestId);
                    
                    switch (genericEvent.getType()) {
                        case "BEGIN":
                            combinedEvent.setStartTime(node.get("startTime").asText());
                            combinedEvent.getDataParts().add(node.get("data").asText());
                            break;
                        case "CONTINUE":
                            combinedEvent.setProcessTime(node.get("processTime").asText());
                            combinedEvent.getDataParts().add(node.get("data").asText());
                            break;
                        case "END":
                            combinedEvent.setEndTime(node.get("endTime").asText());
                            combinedEvent.getDataParts().add(node.get("data").asText());
                            break;
                    }
                    return combinedEvent;
                },
                Materialized.with(Serdes.String(), new JsonSerde<>(CombinedEvent.class))
        );

// Filter to only emit complete combined events (all three fields present)
aggregatedTable.toStream()
        .filter((windowedKey, combined) -> 
                combined.getStartTime() != null && 
                combined.getProcessTime() != null && 
                combined.getEndTime() != null)
        .map((windowedKey, value) -> new KeyValue<>(windowedKey.key(), value))
        .to("COMBINED_RESULT", Produced.with(Serdes.String(), new JsonSerde<>(CombinedEvent.class)));

// Start the application
KafkaStreams streams = new KafkaStreams(builder.build(), yourStreamsConfig);
streams.start();

Key Considerations #

• Correlation Key: Never skip this—without a way to group messages, you’ll end up with incorrect combinations. If your messages don’t have a natural key, generate one using business logic (e.g., user ID + timestamp).
• Window Size: Balance between catching delayed messages and avoiding excessive state storage. Adjust based on your system’s typical message latency.
• Missing Messages: Use leftJoin (in method 1) or add checks (in method 2) to handle cases where one event type is missing. Decide whether to emit partial results or discard them.
• State Management: Kafka Streams stores intermediate state locally or in RocksDB. Ensure you have enough storage and configure state retention policies to avoid bloat.

内容的提问来源于stack exchange，提问作者Anthony