kstreams-simple-branches.java

public static final double SOME_CONSTANT = ...;

// ... //

// define conditions for each branch (edge) of topology
Predicate<String, JsonObject> greaterThan = (String key, JsonObject value) -> {
    double dValue = value.get("my_double_value").getAsDouble();
    return dValue > SOME_CONSTANT;
};

Predicate<String, JsonObject> lessThan = (String key, JsonObject value) -> {
    double dValue = value.get("my_double_value").getAsDouble();
    return dValue < SOME_CONSTANT;
};

Predicate<String, JsonObject> equalTo = (String key, JsonObject value) -> {
    double dValue = value.get("my_double_value").getAsDouble();
    
    // epsilon is an arbitrarily small real number, such as 1e-15
    return Math.abs(dValue - SOME_CONSTANT) < epsilon;
};

Predicate<String, JsonObject>[] conditions = new Predicate<>[] { greaterThan, lessThan, equalTo };
KStream<String, JsonElement>[] branches = inputStream.branch(conditions);

// define a output topic for each branch (edge)
branches[0].to("greater-than-topic");
branches[1].to("less-than-topic");
branches[2].to("equal-to-topic");

KafkaStreams streams = new KafkaStreams(builder.build(), config);
streams.start();