-
Notifications
You must be signed in to change notification settings - Fork 4
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add example for using polyflow and kafka - wip
- Loading branch information
1 parent
c21e80b
commit ee962b9
Showing
27 changed files
with
2,604 additions
and
51 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,65 @@ | ||
| # Use this only in dev environments. It's not intended for production usage. | ||
| version: '3.9' | ||
| services: | ||
| zookeeper: | ||
| image: confluentinc/cp-zookeeper:latest | ||
| environment: | ||
| ZOOKEEPER_CLIENT_PORT: 2181 | ||
| ZOOKEEPER_TICK_TIME: 2000 | ||
| ports: | ||
| - '22181:2181' | ||
|
|
||
| kafka: | ||
| image: confluentinc/cp-kafka:latest | ||
| depends_on: | ||
| - zookeeper | ||
| ports: | ||
| - '9092:9092' | ||
| - '29092:29092' | ||
| environment: | ||
| KAFKA_BROKER_ID: 1 | ||
| KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 | ||
| KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092,PLAINTEXT_HOST://localhost:29092 | ||
| KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT | ||
| KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT | ||
| KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 | ||
|
|
||
| init-kafka: | ||
| image: confluentinc/cp-kafka:latest | ||
| depends_on: | ||
| - kafka | ||
| entrypoint: [ '/bin/bash', '-c' ] | ||
| command: | | ||
| " | ||
| # blocks until kafka is reachable | ||
| echo -e 'Currently available topics:' | ||
| kafka-topics --bootstrap-server kafka:9092 --list | ||
| echo -e 'Creating kafka topics...' | ||
| kafka-topics --bootstrap-server kafka:9092 --create --if-not-exists --topic polyflow-task --replication-factor 1 --partitions 1 | ||
| kafka-topics --bootstrap-server kafka:9092 --create --if-not-exists --topic polyflow-data --replication-factor 1 --partitions 1 | ||
| echo -e 'Resulting topics:' | ||
| kafka-topics --bootstrap-server kafka:9092 --list | ||
| " | ||
| postgres-engine: | ||
| image: postgres:13.2 | ||
| container_name: postgres-engine | ||
| environment: | ||
| POSTGRES_USER: polyflow_user | ||
| POSTGRES_PASSWORD: S3Cr3T! | ||
| POSTGRES_DB: enginedb | ||
| ports: | ||
| - '25433:5432' | ||
|
|
||
| postgres-tasklist: | ||
| image: postgres:13.2 | ||
| container_name: postgres-tasklist | ||
| environment: | ||
| POSTGRES_USER: polyflow_user | ||
| POSTGRES_PASSWORD: S3Cr3T! | ||
| POSTGRES_DB: tasklistdb | ||
| ports: | ||
| - '25432:5432' | ||
|
|
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
16 changes: 16 additions & 0 deletions
16
...ow/src/main/kotlin/io/holunda/polyflow/example/process/approval/kafka/KafkaTopicRouter.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,16 @@ | ||
| package io.holunda.polyflow.example.process.approval.kafka | ||
|
|
||
| import javax.validation.constraints.NotNull | ||
|
|
||
| /** | ||
| * Router to decide where to publish events to. | ||
| */ | ||
| fun interface KafkaTopicRouter { | ||
| /** | ||
| * Retrieves the topic name for given payload type. | ||
| * | ||
| * @param payloadType payload type. | ||
| * @return topic or null, if the event should be dropped. | ||
| */ | ||
| fun topicForPayloadType(payloadType: @NotNull Class<*>): String? | ||
| } |
133 changes: 133 additions & 0 deletions
133
...tlin/io/holunda/polyflow/example/process/approval/kafka/PolyflowAxonKafkaConfiguration.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,133 @@ | ||
| package io.holunda.polyflow.example.process.approval.kafka | ||
|
|
||
| import org.apache.kafka.clients.consumer.ConsumerRecord | ||
| import org.apache.kafka.clients.producer.ProducerRecord | ||
| import org.axonframework.common.AxonConfigurationException | ||
| import org.axonframework.config.EventProcessingConfigurer | ||
| import org.axonframework.eventhandling.EventMessage | ||
| import org.axonframework.eventhandling.PropagatingErrorHandler | ||
| import org.axonframework.extensions.kafka.KafkaProperties | ||
| import org.axonframework.extensions.kafka.autoconfig.KafkaAutoConfiguration | ||
| import org.axonframework.extensions.kafka.eventhandling.DefaultKafkaMessageConverter | ||
| import org.axonframework.extensions.kafka.eventhandling.KafkaMessageConverter | ||
| import org.axonframework.extensions.kafka.eventhandling.producer.KafkaEventPublisher | ||
| import org.axonframework.extensions.kafka.eventhandling.producer.KafkaPublisher | ||
| import org.axonframework.extensions.kafka.eventhandling.producer.ProducerFactory | ||
| import org.axonframework.serialization.Serializer | ||
| import org.springframework.beans.factory.annotation.Qualifier | ||
| import org.springframework.boot.autoconfigure.AutoConfigureBefore | ||
| import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean | ||
| import org.springframework.boot.context.properties.EnableConfigurationProperties | ||
| import org.springframework.context.annotation.Bean | ||
| import org.springframework.context.annotation.Configuration | ||
| import org.springframework.context.annotation.Primary | ||
| import java.util.* | ||
|
|
||
| /** | ||
| * Configure to send polyflow events only if kafka is not disabled (is enabled). | ||
| */ | ||
| @Configuration | ||
| @AutoConfigureBefore( | ||
| KafkaAutoConfiguration::class | ||
| ) // we should run before Axon Kafka autoconfiguration | ||
| @EnableConfigurationProperties(PolyflowAxonKafkaProperties::class) | ||
| class PolyflowAxonKafkaConfiguration { | ||
| @ConditionalOnMissingBean | ||
| @Bean | ||
| fun kafkaTopicRouter(properties: PolyflowAxonKafkaProperties): KafkaTopicRouter { | ||
| return KafkaTopicRouter { payloadType -> | ||
| properties.topics.firstOrNull { it.payloadType.isAssignableFrom(payloadType) }?.topic | ||
| } | ||
| } | ||
|
|
||
| @Bean | ||
| @Primary | ||
| fun routingKafkaMessageConverter( | ||
| @Qualifier("eventSerializer") eventSerializer: Serializer, | ||
| kafkaTopicRouter: KafkaTopicRouter | ||
| ): KafkaMessageConverter<String, ByteArray> { | ||
| val defaultConverter: KafkaMessageConverter<String, ByteArray> = | ||
| DefaultKafkaMessageConverter.builder().serializer(eventSerializer).build() | ||
| return object : KafkaMessageConverter<String, ByteArray> { | ||
| override fun createKafkaMessage( | ||
| eventMessage: EventMessage<*>, | ||
| topic: String | ||
| ): ProducerRecord<String, ByteArray> { | ||
| val topicOverride = kafkaTopicRouter.topicForPayloadType(eventMessage.getPayloadType()) | ||
| return defaultConverter.createKafkaMessage(eventMessage, topicOverride ?: topic) | ||
| } | ||
|
|
||
| override fun readKafkaMessage(consumerRecord: ConsumerRecord<String, ByteArray>): Optional<EventMessage<*>> { | ||
| return defaultConverter.readKafkaMessage(consumerRecord) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Configures a KafkaEventPublisher that sends events to Kafka only if they are routed via kafka event router. | ||
| * | ||
| * @see KafkaAutoConfiguration.kafkaEventPublisher | ||
| */ | ||
| @Bean | ||
| fun routingKafkaEventPublisher( | ||
| kafkaPublisher: KafkaPublisher<String, ByteArray>, | ||
| kafkaProperties: KafkaProperties, | ||
| eventProcessingConfigurer: EventProcessingConfigurer, | ||
| kafkaTopicRouter: KafkaTopicRouter | ||
| ): KafkaEventPublisher<String, ByteArray> { | ||
| val kafkaEventPublisher: KafkaEventPublisher<String, ByteArray> = | ||
| RoutingKafkaEventPublisher.builder<String, ByteArray>() | ||
| .kafkaPublisher(kafkaPublisher) | ||
| .kafkaTopicRouter(kafkaTopicRouter) | ||
| .build() | ||
|
|
||
| /* | ||
| * Register an invocation error handler which re-throws any exception. | ||
| * This will ensure a TrackingEventProcessor to enter the error mode which will retry, and it will ensure the | ||
| * SubscribingEventProcessor to bubble the exception to the callee. For more information see | ||
| * https://docs.axoniq.io/reference-guide/configuring-infrastructure-components/event-processing/event-processors#error-handling | ||
| */ | ||
| // TODO: Check if this still works. Our publisher is no longer in the default processing group, I think. | ||
| eventProcessingConfigurer.registerEventHandler { kafkaEventPublisher } | ||
| .registerListenerInvocationErrorHandler( | ||
| KafkaEventPublisher.DEFAULT_PROCESSING_GROUP | ||
| ) { PropagatingErrorHandler.instance() } | ||
| .assignHandlerTypesMatching( | ||
| KafkaEventPublisher.DEFAULT_PROCESSING_GROUP | ||
| ) { clazz: Class<*> -> | ||
| clazz.isAssignableFrom( | ||
| KafkaEventPublisher::class.java | ||
| ) | ||
| } | ||
| when (val processorMode: KafkaProperties.EventProcessorMode = kafkaProperties.producer.eventProcessorMode) { | ||
| KafkaProperties.EventProcessorMode.SUBSCRIBING -> eventProcessingConfigurer.registerSubscribingEventProcessor(KafkaEventPublisher.DEFAULT_PROCESSING_GROUP) | ||
| KafkaProperties.EventProcessorMode.TRACKING -> eventProcessingConfigurer.registerTrackingEventProcessor(KafkaEventPublisher.DEFAULT_PROCESSING_GROUP) | ||
| KafkaProperties.EventProcessorMode.POOLED_STREAMING -> eventProcessingConfigurer.registerPooledStreamingEventProcessor(KafkaEventPublisher.DEFAULT_PROCESSING_GROUP) | ||
| else -> throw AxonConfigurationException("Unknown Event Processor Mode [$processorMode] detected") | ||
| } | ||
|
|
||
| return kafkaEventPublisher | ||
| } | ||
|
|
||
| // We need to duplicate the bean factory from KafkaAutoConfiguration because there is no way to set `publisherAckTimeout` via configuration properties | ||
| @Bean(destroyMethod = "shutDown") | ||
| fun kafkaAcknowledgingPublisher( | ||
| kafkaProducerFactory: ProducerFactory<String, ByteArray>, | ||
| kafkaMessageConverter: KafkaMessageConverter<String, ByteArray>, | ||
| configuration: org.axonframework.config.Configuration, | ||
| properties: KafkaProperties, | ||
| serializer: Serializer | ||
| ): KafkaPublisher<String, ByteArray> { | ||
| return KafkaPublisher | ||
| .builder<String, ByteArray>() | ||
| .producerFactory(kafkaProducerFactory) | ||
| .messageConverter(kafkaMessageConverter) | ||
| .messageMonitor(configuration.messageMonitor(KafkaPublisher::class.java, "kafkaPublisher")) | ||
| .topicResolver { Optional.of(properties.defaultTopic) } | ||
| .serializer(serializer) | ||
| .publisherAckTimeout( | ||
| properties.producer.properties.getOrDefault("delivery.timeout.ms", "30000").toLong() + 1000 | ||
| ) | ||
| .build() | ||
| } | ||
| } |
21 changes: 21 additions & 0 deletions
21
.../kotlin/io/holunda/polyflow/example/process/approval/kafka/PolyflowAxonKafkaProperties.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,21 @@ | ||
| package io.holunda.polyflow.example.process.approval.kafka | ||
|
|
||
| import org.springframework.boot.context.properties.ConfigurationProperties | ||
| import org.springframework.boot.context.properties.ConstructorBinding | ||
| import org.springframework.boot.context.properties.NestedConfigurationProperty | ||
|
|
||
| @ConfigurationProperties(prefix = "polyflow.axon.kafka") | ||
| @ConstructorBinding | ||
| data class PolyflowAxonKafkaProperties( | ||
| /** | ||
| * List of mappings of payload class to kafka topic name that payload of this class should be directed to. | ||
| */ | ||
| @NestedConfigurationProperty | ||
| val topics: List<PayloadTypeToTopic> | ||
| ) { | ||
| @ConstructorBinding | ||
| data class PayloadTypeToTopic( | ||
| val payloadType: Class<*>, | ||
| val topic: String | ||
| ) | ||
| } |
Oops, something went wrong.