I have java model objects (pojo) that I will send/receive in Kafka. We could create the steams form the pojo structure, it's dynamic...
@Data
@Builder
public class Orders {
// KEY : UUID.toString()
String orderId;
String product;
String orderTimestamp;
// CREATED, SENT, RECEIVED, RETURNED
String status;
}
@Data
@Builder
public class Returns {
// KEY : UUID.toString()
String returnId;
String orderId;
String returnTimestamp;
}
String ORDER_TOPIC = "orders";
String RETURN_TOPIC = "returns";
String CREATE_STREAM_ORDERS = String.format("CREATE STREAM %s (orderId VARCHAR KEY, product VARCHAR, orderTimestamp VARCHAR, status VARCHAR) "
+ " WITH (kafka_topic='%s', partitions=1, value_format='json');", ORDER_TOPIC, ORDER_TOPIC);
String CREATE_STREAM_RETURNS = String.format("CREATE STREAM %s (returnId VARCHAR KEY, orderId VARCHAR, returnTimestamp VARCHAR) "
+ " WITH (kafka_topic='%s', partitions=1, value_format='json');", RETURN_TOPIC, RETURN_TOPIC);
reactorClient
.executeStatement(CREATE_STREAM_ORDERS)
.then(reactorClient.listStreams())
.block();
reactorClient
.executeStatement(CREATE_STREAM_RETURNS)
.then(reactorClient.listStreams())
.block();
.createStream(stream name, pojo class, key to used in the pojo, properties... defaults could be : stream name, 1, json )
reactorClient
.createStream(ORDER_TOPIC, Orders.class, "orderId", of("kafka_topic", ORDER_TOPIC,
"partitions", 1,
"value_format", "json"))
.then(reactorClient.listStreams())
.block();
reactorClient
.createStream(RETURN_TOPIC, Returns.class, "orderId", of("kafka_topic", RETURN_TOPIC,
"partitions", 1,
"value_format", "json"))
.then(reactorClient.listStreams())
.block();