Apache Kafka ストリームの使用

./mvnw clean package -f producer/pom.xml ./mvnw clean package -f aggregator/pom.xml

services: kafka: image: quay.io/strimzi/kafka:latest-kafka-4.1.0 command: [ "sh", "-c", "./bin/kafka-storage.sh format --standalone -t $$(./bin/kafka-storage.sh random-uuid) -c ./config/server.properties && ./bin/kafka-server-start.sh ./config/server.properties --override advertised.listeners=$${KAFKA_ADVERTISED_LISTENERS} --override num.partitions=$${KAFKA_NUM_PARTITIONS} --override group.min.session.timeout.ms=$${KAFKA_GROUP_MIN_SESSION_TIMEOUT_MS}" ] ports: - "9092:9092" environment: LOG_DIR: "/tmp/logs" KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka:9092' KAFKA_NUM_PARTITIONS: 3 KAFKA_GROUP_MIN_SESSION_TIMEOUT_MS: 100 networks: - kafkastreams-network producer: image: quarkus-quickstarts/kafka-streams-producer:1.0 build: context: producer dockerfile: src/main/docker/Dockerfile.${QUARKUS_MODE:-jvm} environment: KAFKA_BOOTSTRAP_SERVERS: kafka:9092 networks: - kafkastreams-network aggregator: image: quarkus-quickstarts/kafka-streams-aggregator:1.0 build: context: aggregator dockerfile: src/main/docker/Dockerfile.${QUARKUS_MODE:-jvm} environment: QUARKUS_KAFKA_STREAMS_BOOTSTRAP_SERVERS: kafka:9092 networks: - kafkastreams-network networks: kafkastreams-network: name: ks

docker run --tty --rm -i --network ks debezium/tooling:latest

kafkacat -b kafka:9092 -C -o beginning -q -t temperatures-aggregated {"avg":34.7,"count":4,"max":49.4,"min":16.8,"stationId":9,"stationName":"Marrakesh","sum":138.8} {"avg":15.7,"count":1,"max":15.7,"min":15.7,"stationId":2,"stationName":"Snowdonia","sum":15.7} {"avg":12.8,"count":7,"max":25.5,"min":-13.8,"stationId":7,"stationName":"Porthsmouth","sum":89.7} ...

package org.acme.kafka.streams.aggregator.streams; import jakarta.enterprise.context.ApplicationScoped; import jakarta.inject.Inject; import org.acme.kafka.streams.aggregator.model.Aggregation; import org.acme.kafka.streams.aggregator.model.WeatherStationData; import org.apache.kafka.streams.KafkaStreams; import org.apache.kafka.streams.errors.InvalidStateStoreException; import org.apache.kafka.streams.state.QueryableStoreTypes; import org.apache.kafka.streams.state.ReadOnlyKeyValueStore; @ApplicationScoped public class InteractiveQueries { @Inject KafkaStreams streams; public GetWeatherStationDataResult getWeatherStationData(int id) { Aggregation result = getWeatherStationStore().get(id); if (result != null) { return GetWeatherStationDataResult.found(WeatherStationData.from(result)); (1) } else { return GetWeatherStationDataResult.notFound(); (2) } } private ReadOnlyKeyValueStore<Integer, Aggregation> getWeatherStationStore() { while (true) { try { return streams.store(StoreQueryParameters .fromNameAndType(TopologyProducer.WEATHER_STATIONS_STORE, QueryableStoreTypes.keyValueStore())); } catch (InvalidStateStoreException e) { // ignore, store not ready yet } } } }

package org.acme.kafka.streams.aggregator.streams; import java.util.Optional; import java.util.OptionalInt; import org.acme.kafka.streams.aggregator.model.WeatherStationData; public class GetWeatherStationDataResult { private static GetWeatherStationDataResult NOT_FOUND = new GetWeatherStationDataResult(null); private final WeatherStationData result; private GetWeatherStationDataResult(WeatherStationData result) { this.result = result; } public static GetWeatherStationDataResult found(WeatherStationData data) { return new GetWeatherStationDataResult(data); } public static GetWeatherStationDataResult notFound() { return NOT_FOUND; } public Optional<WeatherStationData> getResult() { return Optional.ofNullable(result); } }

package org.acme.kafka.streams.aggregator.model; import io.quarkus.runtime.annotations.RegisterForReflection; @RegisterForReflection public class WeatherStationData { public int stationId; public String stationName; public double min = Double.MAX_VALUE; public double max = Double.MIN_VALUE; public int count; public double avg; private WeatherStationData(int stationId, String stationName, double min, double max, int count, double avg) { this.stationId = stationId; this.stationName = stationName; this.min = min; this.max = max; this.count = count; this.avg = avg; } public static WeatherStationData from(Aggregation aggregation) { return new WeatherStationData( aggregation.stationId, aggregation.stationName, aggregation.min, aggregation.max, aggregation.count, aggregation.avg); } }

package org.acme.kafka.streams.aggregator.rest; import java.net.URI; import java.net.URISyntaxException; import java.util.List; import jakarta.enterprise.context.ApplicationScoped; import jakarta.inject.Inject; import jakarta.ws.rs.GET; import jakarta.ws.rs.Path; import jakarta.ws.rs.core.MediaType; import jakarta.ws.rs.core.Response; import jakarta.ws.rs.core.Response.Status; import org.acme.kafka.streams.aggregator.streams.GetWeatherStationDataResult; import org.acme.kafka.streams.aggregator.streams.KafkaStreamsPipeline; @ApplicationScoped @Path("/weather-stations") public class WeatherStationEndpoint { @Inject InteractiveQueries interactiveQueries; @GET @Path("/data/{id}") public Response getWeatherStationData(int id) { GetWeatherStationDataResult result = interactiveQueries.getWeatherStationData(id); if (result.getResult().isPresent()) { (1) return Response.ok(result.getResult().get()).build(); } else { return Response.status(Status.NOT_FOUND.getStatusCode(), "No data found for weather station " + id).build(); } } }

./mvnw clean package -f aggregator/pom.xml docker-compose stop aggregator docker-compose up --build -d

http aggregator:8080/weather-stations/data/1 HTTP/1.1 200 OK Connection: keep-alive Content-Length: 85 Content-Type: application/json Date: Tue, 18 Jun 2019 19:29:16 GMT { "avg": 12.9, "count": 146, "max": 41.0, "min": -25.6, "stationId": 1, "stationName": "Hamburg" }

public GetWeatherStationDataResult getWeatherStationData(int id) { StreamsMetadata metadata = streams.metadataForKey( (1) TopologyProducer.WEATHER_STATIONS_STORE, id, Serdes.Integer().serializer() ); if (metadata == null || metadata == StreamsMetadata.NOT_AVAILABLE) { LOG.warn("Found no metadata for key {}", id); return GetWeatherStationDataResult.notFound(); } else if (metadata.host().equals(host)) { (2) LOG.info("Found data for key {} locally", id); Aggregation result = getWeatherStationStore().get(id); if (result != null) { return GetWeatherStationDataResult.found(WeatherStationData.from(result)); } else { return GetWeatherStationDataResult.notFound(); } } else { (3) LOG.info( "Found data for key {} on remote host {}:{}", id, metadata.host(), metadata.port() ); return GetWeatherStationDataResult.foundRemotely(metadata.host(), metadata.port()); } } public List<PipelineMetadata> getMetaData() { (4) return streams.allMetadataForStore(TopologyProducer.WEATHER_STATIONS_STORE) .stream() .map(m -> new PipelineMetadata( m.hostInfo().host() + ":" + m.hostInfo().port(), m.topicPartitions() .stream() .map(TopicPartition::toString) .collect(Collectors.toSet())) ) .collect(Collectors.toList()); }

package org.acme.kafka.streams.aggregator.streams; import java.util.Optional; import java.util.OptionalInt; import org.acme.kafka.streams.aggregator.model.WeatherStationData; public class GetWeatherStationDataResult { private static GetWeatherStationDataResult NOT_FOUND = new GetWeatherStationDataResult(null, null, null); private final WeatherStationData result; private final String host; private final Integer port; private GetWeatherStationDataResult(WeatherStationData result, String host, Integer port) { this.result = result; this.host = host; this.port = port; } public static GetWeatherStationDataResult found(WeatherStationData data) { return new GetWeatherStationDataResult(data, null, null); } public static GetWeatherStationDataResult foundRemotely(String host, int port) { return new GetWeatherStationDataResult(null, host, port); } public static GetWeatherStationDataResult notFound() { return NOT_FOUND; } public Optional<WeatherStationData> getResult() { return Optional.ofNullable(result); } public Optional<String> getHost() { return Optional.ofNullable(host); } public OptionalInt getPort() { return port != null ? OptionalInt.of(port) : OptionalInt.empty(); } }

package org.acme.kafka.streams.aggregator.streams; import java.util.Set; public class PipelineMetadata { public String host; public Set<String> partitions; public PipelineMetadata(String host, Set<String> partitions) { this.host = host; this.partitions = partitions; } }

package org.acme.kafka.streams.aggregator.rest; import java.net.URI; import java.net.URISyntaxException; import java.util.List; import jakarta.enterprise.context.ApplicationScoped; import jakarta.inject.Inject; import jakarta.ws.rs.Consumes; import jakarta.ws.rs.GET; import jakarta.ws.rs.Path; import jakarta.ws.rs.Produces; import jakarta.ws.rs.core.MediaType; import jakarta.ws.rs.core.Response; import jakarta.ws.rs.core.Response.Status; import org.acme.kafka.streams.aggregator.streams.GetWeatherStationDataResult; import org.acme.kafka.streams.aggregator.streams.KafkaStreamsPipeline; import org.acme.kafka.streams.aggregator.streams.PipelineMetadata; @ApplicationScoped @Path("/weather-stations") public class WeatherStationEndpoint { @Inject InteractiveQueries interactiveQueries; @GET @Path("/data/{id}") @Consumes(MediaType.APPLICATION_JSON) @Produces(MediaType.APPLICATION_JSON) public Response getWeatherStationData(int id) { GetWeatherStationDataResult result = interactiveQueries.getWeatherStationData(id); if (result.getResult().isPresent()) { (1) return Response.ok(result.getResult().get()).build(); } else if (result.getHost().isPresent()) { (2) URI otherUri = getOtherUri(result.getHost().get(), result.getPort().getAsInt(), id); return Response.seeOther(otherUri).build(); } else { (3) return Response.status(Status.NOT_FOUND.getStatusCode(), "No data found for weather station " + id).build(); } } @GET @Path("/meta-data") @Produces(MediaType.APPLICATION_JSON) public List<PipelineMetadata> getMetaData() { (4) return interactiveQueries.getMetaData(); } private URI getOtherUri(String host, int port, int id) { try { return new URI("http://" + host + ":" + port + "/weather-stations/data/" + id); } catch (URISyntaxException e) { throw new RuntimeException(e); } } }

./mvnw clean package -f aggregator/pom.xml docker-compose stop aggregator docker-compose up --build -d --scale aggregator=3

http aggregator:8080/weather-stations/meta-data HTTP/1.1 200 OK Connection: keep-alive Content-Length: 202 Content-Type: application/json Date: Tue, 18 Jun 2019 20:00:23 GMT [ { "host": "2af13fe516a9:8080", "partitions": [ "temperature-values-2" ] }, { "host": "32cc8309611b:8080", "partitions": [ "temperature-values-1" ] }, { "host": "1eb39af8d587:8080", "partitions": [ "temperature-values-0" ] } ]

http 2af13fe516a9:8080/weather-stations/data/1

HTTP/1.1 200 OK Connection: keep-alive Content-Length: 74 Content-Type: application/json Date: Tue, 11 Jun 2019 19:16:31 GMT { "avg": 11.9, "count": 259, "max": 50.0, "min": -30.1, "stationId": 1, "stationName": "Hamburg" }

HTTP/1.1 303 See Other Connection: keep-alive Content-Length: 0 Date: Tue, 18 Jun 2019 20:01:03 GMT Location: http://1eb39af8d587:8080/weather-stations/data/1

http --follow 2af13fe516a9:8080/weather-stations/data/1

./mvnw clean package -f producer/pom.xml -Dnative -Dnative-image.container-runtime=docker ./mvnw clean package -f aggregator/pom.xml -Dnative -Dnative-image.container-runtime=docker

export QUARKUS_MODE=native

CMD ["./application", "-Dquarkus.http.host=0.0.0.0", "-Xmx32m"]

curl -i http://aggregator:8080/q/health HTTP/1.1 503 Service Unavailable content-type: application/json; charset=UTF-8 content-length: 454 { "status": "DOWN", "checks": [ { "name": "Kafka Streams state health check", (1) "status": "DOWN", "data": { "state": "CREATED" } }, { "name": "Kafka Streams topics health check", (2) "status": "DOWN", "data": { "available_topics": "weather-stations,temperature-values", "missing_topics": "hygrometry-values" } } ] }