Skip to content

Commit

Permalink
Spring Boot 3.x (#27)
Browse files Browse the repository at this point in the history
  • Loading branch information
rogervinas authored Dec 1, 2023
1 parent 59e6bb9 commit 8855656
Show file tree
Hide file tree
Showing 5 changed files with 15 additions and 15 deletions.
2 changes: 1 addition & 1 deletion .github/workflows/gradle.yml
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ jobs:
- name: Set up Java
uses: actions/setup-java@v3
with:
java-version: '17'
java-version: '21'
distribution: 'temurin'
- name: Gradle cache
uses: actions/cache@v3
Expand Down
10 changes: 5 additions & 5 deletions README.md
Original file line number Diff line number Diff line change
@@ -1,8 +1,8 @@
![CI](https://github.com/rogervinas/spring-cloud-stream-kafka-step-by-step/actions/workflows/gradle.yml/badge.svg)
![Java](https://img.shields.io/badge/Java-17-blue?labelColor=black)
![Java](https://img.shields.io/badge/Java-21-blue?labelColor=black)
![Kotlin](https://img.shields.io/badge/Kotlin-1.9.21-blue?labelColor=black)
![SpringBoot](https://img.shields.io/badge/SpringBoot-2.7.8-blue?labelColor=black)
![SpringCloud](https://img.shields.io/badge/SpringCloud-2021.0.8-blue?labelColor=black)
![SpringBoot](https://img.shields.io/badge/SpringBoot-3.2.0-blue?labelColor=black)
![SpringCloud](https://img.shields.io/badge/SpringCloud-2023.0.0_RC1-blue?labelColor=black)

# Spring Cloud Stream & Kafka binder step by step

Expand Down Expand Up @@ -244,14 +244,14 @@ When a message is sent to a topic, Kafka chooses randomly the destination partit

This is important on the consumer side, because **chronological order of messages is only guaranteed within the same partition**, so if we need to consume some messages in the order they were produced, we should use the same key for all of them (i.e. for messages of a *user*, we use the *user* id as the message key).

To specify the message key in `MyStreamEventProducer` we can produce `Message<MyEventPayload>` instead of `MyEventPayload` and inform the `KafkaHeaders.MESSAGE_KEY` header:
To specify the message key in `MyStreamEventProducer` we can produce `Message<MyEventPayload>` instead of `MyEventPayload` and inform the `KafkaHeaders.KEY` header:
```kotlin
class MyStreamEventProducer : () -> Flux<Message<MyEventPayload>>, MyEventProducer {
// ...
override fun produce(event: MyEvent) {
val message = MessageBuilder
.withPayload(MyEventPayload(event.text, event.text.length))
.setHeader(KafkaHeaders.MESSAGE_KEY, "key-${event.text.length}")
.setHeader(KafkaHeaders.KEY, "key-${event.text.length}")
.build()
sink.emitNext(message, FAIL_FAST)
}
Expand Down
12 changes: 6 additions & 6 deletions build.gradle.kts
Original file line number Diff line number Diff line change
Expand Up @@ -3,23 +3,23 @@ import org.gradle.api.tasks.testing.logging.TestLogEvent.*
import org.jetbrains.kotlin.gradle.tasks.KotlinCompile

plugins {
id("org.springframework.boot") version "2.7.8"
id("io.spring.dependency-management") version "1.0.15.RELEASE"
id("org.springframework.boot") version "3.2.0"
id("io.spring.dependency-management") version "1.1.4"
kotlin("jvm") version "1.9.21"
kotlin("plugin.spring") version "1.9.21"
}

group = "com.rogervinas"
version = "0.0.1-SNAPSHOT"
java.sourceCompatibility = JavaVersion.VERSION_17
java.targetCompatibility = JavaVersion.VERSION_17
java.sourceCompatibility = JavaVersion.VERSION_21
java.targetCompatibility = JavaVersion.VERSION_21

repositories {
mavenCentral()
maven { url = uri("https://repo.spring.io/milestone") }
}

val springCloudVersion = "2021.0.8"
val springCloudVersion = "2023.0.0-RC1"

dependencies {
implementation("org.springframework.boot:spring-boot-starter-web")
Expand All @@ -45,7 +45,7 @@ dependencyManagement {
tasks.withType<KotlinCompile> {
kotlinOptions {
freeCompilerArgs = listOf("-Xjsr305=strict")
jvmTarget = "17"
jvmTarget = "21"
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ class MyStreamEventProducer : () -> Flux<Message<MyEventPayload>>, MyEventProduc
override fun produce(event: MyEvent) {
val message = MessageBuilder
.withPayload(toPayload(event))
.setHeader(KafkaHeaders.MESSAGE_KEY, toKey(event))
.setHeader(KafkaHeaders.KEY, toKey(event))
.build()
sink.emitNext(message, FAIL_FAST)
}
Expand Down
4 changes: 2 additions & 2 deletions src/main/kotlin/com/rogervinas/stream/shared/MyContainers.kt
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
package com.rogervinas.stream.shared

import javax.annotation.PostConstruct
import javax.annotation.PreDestroy
import jakarta.annotation.PostConstruct
import jakarta.annotation.PreDestroy
import org.springframework.context.annotation.Profile
import org.springframework.stereotype.Component
import org.testcontainers.containers.ComposeContainer
Expand Down

0 comments on commit 8855656

Please sign in to comment.