diff --git a/kafka-e2e-kotlin-sample/customer-service/pom.xml b/kafka-e2e-kotlin-sample/customer-service/pom.xml index 286b636..7da8680 100644 --- a/kafka-e2e-kotlin-sample/customer-service/pom.xml +++ b/kafka-e2e-kotlin-sample/customer-service/pom.xml @@ -3,25 +3,38 @@ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> 4.0.0 - - io.spring.cloud.stream.sample - spring-cloud-stream-samples-parent - 0.0.1-SNAPSHOT - ../.. - - customer-service 0.0.1-SNAPSHOT customer-service Customer Service + + org.springframework.boot + spring-boot-starter-parent + 2.2.0.BUILD-SNAPSHOT + + + 1.8 1.3.31 1.8.2 5.2.1 + Hoxton.BUILD-SNAPSHOT + + + + org.springframework.cloud + spring-cloud-dependencies + ${spring-cloud.version} + pom + import + + + + io.confluent @@ -94,6 +107,18 @@ spring-cloud-stream-test-support test + + org.springframework.boot + spring-boot-starter-actuator + + + org.springframework.boot + spring-boot-starter + + + org.springframework.boot + spring-boot-starter-web + @@ -145,6 +170,25 @@ + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + true + + + false + + + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + false + + confluent https://packages.confluent.io/maven/ diff --git a/kafka-e2e-kotlin-sample/customer-service/src/main/kotlin/kafka/e2e/customer/configuration/SchemaRegistryConfiguration.kt b/kafka-e2e-kotlin-sample/customer-service/src/main/kotlin/kafka/e2e/customer/configuration/SchemaRegistryConfiguration.kt deleted file mode 100644 index 66ae9f2..0000000 --- a/kafka-e2e-kotlin-sample/customer-service/src/main/kotlin/kafka/e2e/customer/configuration/SchemaRegistryConfiguration.kt +++ /dev/null @@ -1,38 +0,0 @@ -/* - * Copyright 2019 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package kafka.e2e.customer.configuration - -import org.springframework.beans.factory.annotation.Value -import org.springframework.context.annotation.Configuration -import org.springframework.cloud.stream.schema.client.ConfluentSchemaRegistryClient -import org.springframework.cloud.stream.schema.client.SchemaRegistryClient -import org.springframework.context.annotation.Bean - -/** - * @author José A. Íñigo - */ -@Configuration -class SchemaRegistryConfiguration { - - @Bean - fun schemaRegistryClient(@Value("\${spring.cloud.stream.schema-registry-client.endpoint}") endpoint: String): SchemaRegistryClient { - val client = ConfluentSchemaRegistryClient() - client.setEndpoint(endpoint) - return client - } - -} diff --git a/kafka-e2e-kotlin-sample/customer-service/src/main/resources/application.yml b/kafka-e2e-kotlin-sample/customer-service/src/main/resources/application.yml index 29d1c10..e90a49a 100644 --- a/kafka-e2e-kotlin-sample/customer-service/src/main/resources/application.yml +++ b/kafka-e2e-kotlin-sample/customer-service/src/main/resources/application.yml @@ -3,8 +3,8 @@ spring: name: customer-service cloud: stream: - schema-registry-client: - endpoint: http://localhost:8081 +# schema-registry-client: +# endpoint: http://localhost:8081 kafka: bindings: output: @@ -19,7 +19,7 @@ spring: strategy: io.confluent.kafka.serializers.subject.RecordNameStrategy schema: registry: - url: ${spring.cloud.stream.schema-registry-client.endpoint} + url: http://localhost:8081 bindings: output: destination: customer diff --git a/kafka-e2e-kotlin-sample/order-service/pom.xml b/kafka-e2e-kotlin-sample/order-service/pom.xml index 44c9783..2cf1eaa 100644 --- a/kafka-e2e-kotlin-sample/order-service/pom.xml +++ b/kafka-e2e-kotlin-sample/order-service/pom.xml @@ -3,25 +3,38 @@ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> 4.0.0 - - io.spring.cloud.stream.sample - spring-cloud-stream-samples-parent - 0.0.1-SNAPSHOT - ../.. - - order-service 0.0.1-SNAPSHOT order-service Order Service + + org.springframework.boot + spring-boot-starter-parent + 2.2.0.BUILD-SNAPSHOT + + + 1.8 1.3.31 1.8.2 5.2.1 + Hoxton.BUILD-SNAPSHOT + + + + org.springframework.cloud + spring-cloud-dependencies + ${spring-cloud.version} + pom + import + + + + io.confluent @@ -145,6 +158,25 @@ + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + true + + + false + + + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + false + + confluent https://packages.confluent.io/maven/ diff --git a/kafka-e2e-kotlin-sample/order-service/src/main/kotlin/kafka/e2e/order/configuration/SchemaRegistryConfiguration.kt b/kafka-e2e-kotlin-sample/order-service/src/main/kotlin/kafka/e2e/order/configuration/SchemaRegistryConfiguration.kt deleted file mode 100644 index 21124c1..0000000 --- a/kafka-e2e-kotlin-sample/order-service/src/main/kotlin/kafka/e2e/order/configuration/SchemaRegistryConfiguration.kt +++ /dev/null @@ -1,38 +0,0 @@ -/* - * Copyright 2019 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package kafka.e2e.order.configuration - -import org.springframework.beans.factory.annotation.Value -import org.springframework.context.annotation.Configuration -import org.springframework.cloud.stream.schema.client.ConfluentSchemaRegistryClient -import org.springframework.cloud.stream.schema.client.SchemaRegistryClient -import org.springframework.context.annotation.Bean - -/** - * @author José A. Íñigo - */ -@Configuration -class SchemaRegistryConfiguration { - - @Bean - fun schemaRegistryClient(@Value("\${spring.cloud.stream.schema-registry-client.endpoint}") endpoint: String): SchemaRegistryClient { - val client = ConfluentSchemaRegistryClient() - client.setEndpoint(endpoint) - return client - } - -} diff --git a/kafka-e2e-kotlin-sample/order-service/src/main/resources/application.yml b/kafka-e2e-kotlin-sample/order-service/src/main/resources/application.yml index af0603e..25455ae 100644 --- a/kafka-e2e-kotlin-sample/order-service/src/main/resources/application.yml +++ b/kafka-e2e-kotlin-sample/order-service/src/main/resources/application.yml @@ -3,8 +3,8 @@ spring: name: order-service cloud: stream: - schema-registry-client: - endpoint: http://localhost:8081 +# schema-registry-client: +# endpoint: http://localhost:8081 kafka: bindings: output: @@ -19,7 +19,7 @@ spring: strategy: io.confluent.kafka.serializers.subject.RecordNameStrategy schema: registry: - url: ${spring.cloud.stream.schema-registry-client.endpoint} + url: http://localhost:8081 bindings: output: destination: order diff --git a/kafka-e2e-kotlin-sample/shipping-service/pom.xml b/kafka-e2e-kotlin-sample/shipping-service/pom.xml index 445537d..c41a6ab 100644 --- a/kafka-e2e-kotlin-sample/shipping-service/pom.xml +++ b/kafka-e2e-kotlin-sample/shipping-service/pom.xml @@ -3,25 +3,38 @@ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> 4.0.0 - - io.spring.cloud.stream.sample - spring-cloud-stream-samples-parent - 0.0.1-SNAPSHOT - ../.. - - shipping-service 0.0.1-SNAPSHOT shipping-service Shipping Service + + org.springframework.boot + spring-boot-starter-parent + 2.2.0.BUILD-SNAPSHOT + + + 1.8 1.3.31 1.8.2 5.2.1 + Hoxton.BUILD-SNAPSHOT + + + + org.springframework.cloud + spring-cloud-dependencies + ${spring-cloud.version} + pom + import + + + + io.confluent @@ -94,6 +107,18 @@ spring-cloud-stream-test-support test + + org.springframework.boot + spring-boot-starter-actuator + + + org.springframework.boot + spring-boot-starter + + + org.springframework.boot + spring-boot-starter-web + @@ -145,6 +170,25 @@ + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot-local + + true + + + false + + + + spring-milestones + Spring Milestones + https://repo.spring.io/libs-milestone-local + + false + + confluent https://packages.confluent.io/maven/ diff --git a/kafka-e2e-kotlin-sample/shipping-service/src/main/kotlin/kafka/e2e/shipping/configuration/SchemaRegistryConfiguration.kt b/kafka-e2e-kotlin-sample/shipping-service/src/main/kotlin/kafka/e2e/shipping/configuration/SchemaRegistryConfiguration.kt deleted file mode 100644 index 88bedd0..0000000 --- a/kafka-e2e-kotlin-sample/shipping-service/src/main/kotlin/kafka/e2e/shipping/configuration/SchemaRegistryConfiguration.kt +++ /dev/null @@ -1,46 +0,0 @@ -/* - * Copyright 2019 the original author or authors. - * - * Licensed under the Apache License, Version 2.0 (the "License"); - * you may not use this file except in compliance with the License. - * You may obtain a copy of the License at - * - * https://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package kafka.e2e.shipping.configuration - -import org.springframework.beans.factory.annotation.Value -import org.springframework.cloud.stream.schema.avro.AvroSchemaMessageConverter -import org.springframework.cloud.stream.schema.client.ConfluentSchemaRegistryClient -import org.springframework.cloud.stream.schema.client.SchemaRegistryClient -import org.springframework.context.annotation.Bean -import org.springframework.context.annotation.Configuration -import org.springframework.messaging.converter.MessageConverter -import org.springframework.util.MimeType - -/** - * @author José A. Íñigo - */ -@Configuration -class SchemaRegistryConfiguration { - - @Bean - fun schemaRegistryClient(@Value("\${spring.cloud.stream.schema-registry-client.endpoint}") endpoint: String): SchemaRegistryClient { - val client = ConfluentSchemaRegistryClient() - client.setEndpoint(endpoint) - return client - } - - @Bean - fun messageConverter(): MessageConverter { - return AvroSchemaMessageConverter(MimeType.valueOf("avro/bytes")) - } - -} diff --git a/kafka-e2e-kotlin-sample/shipping-service/src/main/kotlin/kafka/e2e/shipping/stream/ShippingKStreamConfiguration.kt b/kafka-e2e-kotlin-sample/shipping-service/src/main/kotlin/kafka/e2e/shipping/stream/ShippingKStreamConfiguration.kt index dff7dd7..a06773a 100644 --- a/kafka-e2e-kotlin-sample/shipping-service/src/main/kotlin/kafka/e2e/shipping/stream/ShippingKStreamConfiguration.kt +++ b/kafka-e2e-kotlin-sample/shipping-service/src/main/kotlin/kafka/e2e/shipping/stream/ShippingKStreamConfiguration.kt @@ -56,10 +56,8 @@ class ShippingKStreamConfiguration { val stateStore: Materialized> = Materialized.`as`>("customer-store") - .withKeySerde(intSerde) - .withValueSerde(customerSerde) - val customerTable: KTable = input.groupByKey(Serialized.with(intSerde, customerSerde)) + val customerTable: KTable = input.groupByKey() .reduce({ _, y -> y }, stateStore) return (orderEvent.filter { _, value -> value.schema.name == "OrderCreatedEvent" } diff --git a/kafka-e2e-kotlin-sample/shipping-service/src/main/resources/application.yml b/kafka-e2e-kotlin-sample/shipping-service/src/main/resources/application.yml index 7f0331a..4d4e68b 100644 --- a/kafka-e2e-kotlin-sample/shipping-service/src/main/resources/application.yml +++ b/kafka-e2e-kotlin-sample/shipping-service/src/main/resources/application.yml @@ -3,20 +3,15 @@ spring: name: shipping-service cloud: stream: - schema-registry-client: - endpoint: http://localhost:8081 kafka: streams: binder: configuration: application: id: shipping-service - default: - key: - serde: org.apache.kafka.common.serialization.Serdes$IntegerSerde schema: registry: - url: ${spring.cloud.stream.schema-registry-client.endpoint} + url: http://localhost:8081 value: subject: name: