1 package de.trion.microservices.validateorder;
4 import de.trion.microservices.avro.Order;
5 import de.trion.microservices.avro.OrderState;
6 import de.trion.microservices.avro.OrderValidation;
7 import static de.trion.microservices.avro.OrderValidationResult.FAIL;
8 import static de.trion.microservices.avro.OrderValidationResult.PASS;
9 import static de.trion.microservices.avro.OrderValidationType.ORDER_DETAILS_CHECK;
10 import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde;
11 import java.util.LinkedList;
12 import java.util.List;
13 import java.util.Properties;
14 import javax.annotation.PostConstruct;
15 import javax.annotation.PreDestroy;
16 import org.apache.kafka.common.serialization.Serdes;
17 import org.apache.kafka.streams.KafkaStreams;
18 import org.apache.kafka.streams.StreamsBuilder;
19 import org.apache.kafka.streams.Topology;
20 import org.slf4j.Logger;
21 import org.slf4j.LoggerFactory;
22 import org.springframework.stereotype.Component;
26 public class ValidateOrderService
28 final static Logger LOG = LoggerFactory.getLogger(ValidateOrderService.class);
30 private final String orders;
31 private final String validation;
32 private final KafkaStreams streams;
35 public ValidateOrderService(ApplicationProperties config)
37 orders = config.ordersTopic;
38 validation = config.validationTopic;
40 Properties properties = new Properties();
41 properties.put("bootstrap.servers", config.bootstrapServers);
42 properties.put("application.id", "validate-order");
43 properties.put("schema.registry.url", config.schemaRegistryUrl);
44 properties.put("default.key.serde", Serdes.String().getClass());
45 properties.put("default.value.serde", SpecificAvroSerde.class);
47 StreamsBuilder builder = new StreamsBuilder();
49 .<String,Order>stream(orders)
50 .filter((id, order) -> order.getState() == OrderState.CREATED)
53 List<CharSequence> messages = new LinkedList<>();
55 if (order.getCustomerId() < 1)
56 messages.add("Customer-ID must be greater than 0");
57 if (order.getProductId() < 1)
58 messages.add("Product-ID must be greater than 0");
59 if (order.getQuantity() < 1)
60 messages.add("The ordered quantity must be greater than 0");
65 .setOrderId(order.getId())
66 .setCheckType(ORDER_DETAILS_CHECK)
67 .setValidationResult(messages.isEmpty() ? PASS : FAIL)
68 .setMessages(messages)
73 Topology topology = builder.build();
74 streams = new KafkaStreams(topology, properties);
75 streams.setUncaughtExceptionHandler((Thread t, Throwable e) ->
77 LOG.error("Unexpected error in thread {}: {}", t, e.toString());
84 LOG.error("Could not close KafkaStreams!", ex);