From 17270e0e97a4fedcbd09ffa057c5cfcfa6074b1e Mon Sep 17 00:00:00 2001 From: Florian Brons Date: Tue, 15 Nov 2022 11:38:00 +0100 Subject: [PATCH 1/4] #112 Do not lock all available events --- README.md | 4 +++ nakadi-producer-loadtest/pom.xml | 4 +-- nakadi-producer-spring-boot-starter/pom.xml | 4 +-- .../NakadiProducerAutoConfiguration.java | 3 +- .../eventlog/impl/EventLogRepositoryImpl.java | 17 +++++++-- .../EventLockSizeConfiguredIT.java | 36 +++++++++++++++++++ .../EventLockSizeDefaultIT.java | 30 ++++++++++++++++ nakadi-producer/pom.xml | 4 +-- .../eventlog/impl/EventLogRepository.java | 2 +- .../impl/EventTransmissionService.java | 12 ++++--- .../impl/EventTransmissionServiceTest.java | 2 +- pom.xml | 4 +-- 12 files changed, 105 insertions(+), 17 deletions(-) create mode 100644 nakadi-producer-spring-boot-starter/src/test/java/org/zalando/nakadiproducer/EventLockSizeConfiguredIT.java create mode 100644 nakadi-producer-spring-boot-starter/src/test/java/org/zalando/nakadiproducer/EventLockSizeDefaultIT.java diff --git a/README.md b/README.md index 530bf112..051bb7ca 100644 --- a/README.md +++ b/README.md @@ -335,10 +335,14 @@ until the lock expires. The default is currently 600 seconds but might change in buffer is included. During the last x seconds before the expiration of the lock the events are not considered for transmission. The default is currently 60 seconds but might change in future releases. +* **lock-size**: Defines the amount of events which are loaded into memory and published in one run. + By default, all events are loaded into memory. + ```yaml nakadi-producer: lock-duration: 600 lock-duration-buffer: 60 + lock-size: 5000 ``` ## Contributing diff --git a/nakadi-producer-loadtest/pom.xml b/nakadi-producer-loadtest/pom.xml index b32962a2..a33fc775 100644 --- a/nakadi-producer-loadtest/pom.xml +++ b/nakadi-producer-loadtest/pom.xml @@ -11,7 +11,7 @@ org.zalando nakadi-producer-reactor - 20.3.1 + 20.3.2 @@ -62,4 +62,4 @@ - \ No newline at end of file + diff --git a/nakadi-producer-spring-boot-starter/pom.xml b/nakadi-producer-spring-boot-starter/pom.xml index ead3a959..a79813f0 100644 --- a/nakadi-producer-spring-boot-starter/pom.xml +++ b/nakadi-producer-spring-boot-starter/pom.xml @@ -10,7 +10,7 @@ org.zalando nakadi-producer-reactor - 20.3.1 + 20.3.2 nakadi-producer-spring-boot-starter @@ -194,4 +194,4 @@ - \ No newline at end of file + diff --git a/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/NakadiProducerAutoConfiguration.java b/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/NakadiProducerAutoConfiguration.java index 6344d734..e15cd379 100644 --- a/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/NakadiProducerAutoConfiguration.java +++ b/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/NakadiProducerAutoConfiguration.java @@ -153,10 +153,11 @@ public EventTransmissionService eventTransmissionService( EventLogRepository eventLogRepository, NakadiPublishingClient nakadiPublishingClient, ObjectMapper objectMapper, + @Value("${nakadi-producer.lock-size:0}") int lockSize, @Value("${nakadi-producer.lock-duration:600}") int lockDuration, @Value("${nakadi-producer.lock-duration-buffer:60}") int lockDurationBuffer) { return new EventTransmissionService( - eventLogRepository, nakadiPublishingClient, objectMapper, lockDuration, lockDurationBuffer); + eventLogRepository, nakadiPublishingClient, objectMapper, lockSize, lockDuration, lockDurationBuffer); } @Bean diff --git a/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepositoryImpl.java b/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepositoryImpl.java index b8af1121..a1a07fb0 100644 --- a/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepositoryImpl.java +++ b/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepositoryImpl.java @@ -32,13 +32,26 @@ public Collection findByLockedByAndLockedUntilGreaterThan(String locke } @Override - public void lockSomeMessages(String lockId, Instant now, Instant lockExpires) { + public void lockSomeMessages(String lockId, int lockSize, Instant now, Instant lockExpires) { Map namedParameterMap = new HashMap<>(); namedParameterMap.put("lockId", lockId); namedParameterMap.put("now", toSqlTimestamp(now)); namedParameterMap.put("lockExpires", toSqlTimestamp(lockExpires)); + + StringBuilder optionalLockSizeClause = new StringBuilder(); + if (lockSize > 0) { + optionalLockSizeClause.append("LIMIT :lockSize"); + namedParameterMap.put("lockSize", lockSize); + } + jdbcTemplate.update( - "UPDATE nakadi_events.event_log SET locked_by = :lockId, locked_until = :lockExpires where locked_until is null or locked_until < :now", + "UPDATE nakadi_events.event_log " + + "SET locked_by = :lockId, locked_until = :lockExpires " + + "WHERE id IN (SELECT id " + + " FROM nakadi_events.event_log " + + " WHERE locked_until IS null OR locked_until < :now " + + optionalLockSizeClause + + " FOR UPDATE SKIP LOCKED) ", namedParameterMap ); } diff --git a/nakadi-producer-spring-boot-starter/src/test/java/org/zalando/nakadiproducer/EventLockSizeConfiguredIT.java b/nakadi-producer-spring-boot-starter/src/test/java/org/zalando/nakadiproducer/EventLockSizeConfiguredIT.java new file mode 100644 index 00000000..ba616d86 --- /dev/null +++ b/nakadi-producer-spring-boot-starter/src/test/java/org/zalando/nakadiproducer/EventLockSizeConfiguredIT.java @@ -0,0 +1,36 @@ +package org.zalando.nakadiproducer; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.hasSize; + +import org.junit.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.zalando.nakadiproducer.eventlog.EventLogWriter; +import org.zalando.nakadiproducer.transmission.impl.EventTransmissionService; +import org.zalando.nakadiproducer.util.Fixture; + +@SpringBootTest( + properties = {"nakadi-producer.lock-size=3"} +) +public class EventLockSizeConfiguredIT extends BaseMockedExternalCommunicationIT { + + @Autowired + private EventLogWriter eventLogWriter; + + @Autowired + private EventTransmissionService eventTransmissionService; + + @Test + public void eventLockSizeIsRespected() { + + for (int i = 1; i <= 8; i++) { + eventLogWriter.fireBusinessEvent( "myEventType", Fixture.mockPayload(i, "code123")); + } + + assertThat(eventTransmissionService.lockSomeEvents(), hasSize(3)); + assertThat(eventTransmissionService.lockSomeEvents(), hasSize(3)); + assertThat(eventTransmissionService.lockSomeEvents(), hasSize(2)); + } + +} diff --git a/nakadi-producer-spring-boot-starter/src/test/java/org/zalando/nakadiproducer/EventLockSizeDefaultIT.java b/nakadi-producer-spring-boot-starter/src/test/java/org/zalando/nakadiproducer/EventLockSizeDefaultIT.java new file mode 100644 index 00000000..08ddfc8a --- /dev/null +++ b/nakadi-producer-spring-boot-starter/src/test/java/org/zalando/nakadiproducer/EventLockSizeDefaultIT.java @@ -0,0 +1,30 @@ +package org.zalando.nakadiproducer; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.hasSize; + +import org.junit.Test; +import org.springframework.beans.factory.annotation.Autowired; +import org.zalando.nakadiproducer.eventlog.EventLogWriter; +import org.zalando.nakadiproducer.transmission.impl.EventTransmissionService; +import org.zalando.nakadiproducer.util.Fixture; + +public class EventLockSizeDefaultIT extends BaseMockedExternalCommunicationIT { + + @Autowired + private EventLogWriter eventLogWriter; + + @Autowired + private EventTransmissionService eventTransmissionService; + + @Test + public void defaultEventLockSizeIsUsed() { + + for (int i = 1; i <= 8; i++) { + eventLogWriter.fireBusinessEvent("myEventType", Fixture.mockPayload(i, "code123")); + } + + assertThat(eventTransmissionService.lockSomeEvents(), hasSize(8)); + } + +} diff --git a/nakadi-producer/pom.xml b/nakadi-producer/pom.xml index 2c433840..fbe6acfc 100644 --- a/nakadi-producer/pom.xml +++ b/nakadi-producer/pom.xml @@ -10,7 +10,7 @@ org.zalando nakadi-producer-reactor - 20.3.1 + 20.3.2 nakadi-producer @@ -161,4 +161,4 @@ - \ No newline at end of file + diff --git a/nakadi-producer/src/main/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepository.java b/nakadi-producer/src/main/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepository.java index 937ffdc3..d1a7ca9f 100644 --- a/nakadi-producer/src/main/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepository.java +++ b/nakadi-producer/src/main/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepository.java @@ -6,7 +6,7 @@ public interface EventLogRepository { Collection findByLockedByAndLockedUntilGreaterThan(String lockedBy, Instant lockedUntil); - void lockSomeMessages(String lockId, Instant now, Instant lockExpires); + void lockSomeMessages(String lockId, int lockSize, Instant now, Instant lockExpires); void delete(EventLog eventLog); diff --git a/nakadi-producer/src/main/java/org/zalando/nakadiproducer/transmission/impl/EventTransmissionService.java b/nakadi-producer/src/main/java/org/zalando/nakadiproducer/transmission/impl/EventTransmissionService.java index a4ef056f..6c257ca5 100644 --- a/nakadi-producer/src/main/java/org/zalando/nakadiproducer/transmission/impl/EventTransmissionService.java +++ b/nakadi-producer/src/main/java/org/zalando/nakadiproducer/transmission/impl/EventTransmissionService.java @@ -31,16 +31,19 @@ public class EventTransmissionService { private final EventLogRepository eventLogRepository; private final NakadiPublishingClient nakadiPublishingClient; private final ObjectMapper objectMapper; + private final int lockSize; private final int lockDuration; private final int lockDurationBuffer; private Clock clock = Clock.systemDefaultZone(); - public EventTransmissionService(EventLogRepository eventLogRepository, NakadiPublishingClient nakadiPublishingClient, ObjectMapper objectMapper, - int lockDuration, int lockDurationBuffer) { + public EventTransmissionService(EventLogRepository eventLogRepository, + NakadiPublishingClient nakadiPublishingClient, ObjectMapper objectMapper, int lockSize, + int lockDuration, int lockDurationBuffer) { this.eventLogRepository = eventLogRepository; this.nakadiPublishingClient = nakadiPublishingClient; this.objectMapper = objectMapper; + this.lockSize = lockSize; this.lockDuration = lockDuration; this.lockDurationBuffer = lockDurationBuffer; } @@ -48,8 +51,9 @@ public EventTransmissionService(EventLogRepository eventLogRepository, NakadiPub @Transactional public Collection lockSomeEvents() { String lockId = UUID.randomUUID().toString(); - log.debug("Locking events for replication with lockId {} for {} seconds", lockId, lockDuration); - eventLogRepository.lockSomeMessages(lockId, now(), now().plus(lockDuration, SECONDS)); + log.debug("Locking {} events for replication with lockId {} for {} seconds", lockSize, + lockId, lockDuration); + eventLogRepository.lockSomeMessages(lockId, lockSize, now(), now().plus(lockDuration, SECONDS)); return eventLogRepository.findByLockedByAndLockedUntilGreaterThan(lockId, now()); } diff --git a/nakadi-producer/src/test/java/org/zalando/nakadiproducer/transmission/impl/EventTransmissionServiceTest.java b/nakadi-producer/src/test/java/org/zalando/nakadiproducer/transmission/impl/EventTransmissionServiceTest.java index fecfbf67..b51cd7c3 100644 --- a/nakadi-producer/src/test/java/org/zalando/nakadiproducer/transmission/impl/EventTransmissionServiceTest.java +++ b/nakadi-producer/src/test/java/org/zalando/nakadiproducer/transmission/impl/EventTransmissionServiceTest.java @@ -52,7 +52,7 @@ public void setUp() { repo = mock(EventLogRepository.class); publishingClient = spy(new MockNakadiPublishingClient()); mapper = spy(new ObjectMapper()); - service = new EventTransmissionService(repo, publishingClient, mapper, 600, 60); + service = new EventTransmissionService(repo, publishingClient, mapper, 0, 600, 60); } @Test diff --git a/pom.xml b/pom.xml index 4556ac4a..3b55880d 100644 --- a/pom.xml +++ b/pom.xml @@ -15,7 +15,7 @@ nakadi-producer-reactor org.zalando - 20.3.1 + 20.3.2 pom Nakadi Event Producer Reactor @@ -78,4 +78,4 @@ - \ No newline at end of file + From 6c31f4cb7f5daf69d1f2eb4ddd6287eda2012b9b Mon Sep 17 00:00:00 2001 From: Florian Brons Date: Tue, 15 Nov 2022 14:15:28 +0100 Subject: [PATCH 2/4] #112 Do not lock all available events --- README.md | 4 ++-- .../NakadiProducerAutoConfiguration.java | 8 ++++---- .../eventlog/impl/EventLogRepositoryImpl.java | 7 +++++-- .../eventlog/impl/EventLogRepository.java | 2 +- .../transmission/impl/EventTransmissionService.java | 12 ++++-------- .../impl/EventTransmissionServiceTest.java | 2 +- 6 files changed, 17 insertions(+), 18 deletions(-) diff --git a/README.md b/README.md index 051bb7ca..16308b63 100644 --- a/README.md +++ b/README.md @@ -335,8 +335,8 @@ until the lock expires. The default is currently 600 seconds but might change in buffer is included. During the last x seconds before the expiration of the lock the events are not considered for transmission. The default is currently 60 seconds but might change in future releases. -* **lock-size**: Defines the amount of events which are loaded into memory and published in one run. - By default, all events are loaded into memory. +* **lock-size**: Defines the maximum amount of events which are loaded into memory and published in one run + (in one submission per event type). By default, all events are loaded into memory. ```yaml nakadi-producer: diff --git a/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/NakadiProducerAutoConfiguration.java b/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/NakadiProducerAutoConfiguration.java index e15cd379..6f35e23c 100644 --- a/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/NakadiProducerAutoConfiguration.java +++ b/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/NakadiProducerAutoConfiguration.java @@ -133,8 +133,9 @@ public EventLogWriter eventLogWriter(EventLogRepository eventLogRepository, Obje } @Bean - public EventLogRepository eventLogRepository(NamedParameterJdbcTemplate namedParameterJdbcTemplate) { - return new EventLogRepositoryImpl(namedParameterJdbcTemplate); + public EventLogRepository eventLogRepository(NamedParameterJdbcTemplate namedParameterJdbcTemplate, + @Value("${nakadi-producer.lock-size:0}") int lockSize) { + return new EventLogRepositoryImpl(namedParameterJdbcTemplate, lockSize); } @Bean @@ -153,11 +154,10 @@ public EventTransmissionService eventTransmissionService( EventLogRepository eventLogRepository, NakadiPublishingClient nakadiPublishingClient, ObjectMapper objectMapper, - @Value("${nakadi-producer.lock-size:0}") int lockSize, @Value("${nakadi-producer.lock-duration:600}") int lockDuration, @Value("${nakadi-producer.lock-duration-buffer:60}") int lockDurationBuffer) { return new EventTransmissionService( - eventLogRepository, nakadiPublishingClient, objectMapper, lockSize, lockDuration, lockDurationBuffer); + eventLogRepository, nakadiPublishingClient, objectMapper, lockDuration, lockDurationBuffer); } @Bean diff --git a/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepositoryImpl.java b/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepositoryImpl.java index a1a07fb0..bbb62b40 100644 --- a/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepositoryImpl.java +++ b/nakadi-producer-spring-boot-starter/src/main/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepositoryImpl.java @@ -13,10 +13,13 @@ import org.springframework.jdbc.support.GeneratedKeyHolder; public class EventLogRepositoryImpl implements EventLogRepository { + private NamedParameterJdbcTemplate jdbcTemplate; + private int lockSize; - public EventLogRepositoryImpl(NamedParameterJdbcTemplate jdbcTemplate) { + public EventLogRepositoryImpl(NamedParameterJdbcTemplate jdbcTemplate, int lockSize) { this.jdbcTemplate = jdbcTemplate; + this.lockSize = lockSize; } @Override @@ -32,7 +35,7 @@ public Collection findByLockedByAndLockedUntilGreaterThan(String locke } @Override - public void lockSomeMessages(String lockId, int lockSize, Instant now, Instant lockExpires) { + public void lockSomeMessages(String lockId, Instant now, Instant lockExpires) { Map namedParameterMap = new HashMap<>(); namedParameterMap.put("lockId", lockId); namedParameterMap.put("now", toSqlTimestamp(now)); diff --git a/nakadi-producer/src/main/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepository.java b/nakadi-producer/src/main/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepository.java index d1a7ca9f..937ffdc3 100644 --- a/nakadi-producer/src/main/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepository.java +++ b/nakadi-producer/src/main/java/org/zalando/nakadiproducer/eventlog/impl/EventLogRepository.java @@ -6,7 +6,7 @@ public interface EventLogRepository { Collection findByLockedByAndLockedUntilGreaterThan(String lockedBy, Instant lockedUntil); - void lockSomeMessages(String lockId, int lockSize, Instant now, Instant lockExpires); + void lockSomeMessages(String lockId, Instant now, Instant lockExpires); void delete(EventLog eventLog); diff --git a/nakadi-producer/src/main/java/org/zalando/nakadiproducer/transmission/impl/EventTransmissionService.java b/nakadi-producer/src/main/java/org/zalando/nakadiproducer/transmission/impl/EventTransmissionService.java index 6c257ca5..a4ef056f 100644 --- a/nakadi-producer/src/main/java/org/zalando/nakadiproducer/transmission/impl/EventTransmissionService.java +++ b/nakadi-producer/src/main/java/org/zalando/nakadiproducer/transmission/impl/EventTransmissionService.java @@ -31,19 +31,16 @@ public class EventTransmissionService { private final EventLogRepository eventLogRepository; private final NakadiPublishingClient nakadiPublishingClient; private final ObjectMapper objectMapper; - private final int lockSize; private final int lockDuration; private final int lockDurationBuffer; private Clock clock = Clock.systemDefaultZone(); - public EventTransmissionService(EventLogRepository eventLogRepository, - NakadiPublishingClient nakadiPublishingClient, ObjectMapper objectMapper, int lockSize, - int lockDuration, int lockDurationBuffer) { + public EventTransmissionService(EventLogRepository eventLogRepository, NakadiPublishingClient nakadiPublishingClient, ObjectMapper objectMapper, + int lockDuration, int lockDurationBuffer) { this.eventLogRepository = eventLogRepository; this.nakadiPublishingClient = nakadiPublishingClient; this.objectMapper = objectMapper; - this.lockSize = lockSize; this.lockDuration = lockDuration; this.lockDurationBuffer = lockDurationBuffer; } @@ -51,9 +48,8 @@ public EventTransmissionService(EventLogRepository eventLogRepository, @Transactional public Collection lockSomeEvents() { String lockId = UUID.randomUUID().toString(); - log.debug("Locking {} events for replication with lockId {} for {} seconds", lockSize, - lockId, lockDuration); - eventLogRepository.lockSomeMessages(lockId, lockSize, now(), now().plus(lockDuration, SECONDS)); + log.debug("Locking events for replication with lockId {} for {} seconds", lockId, lockDuration); + eventLogRepository.lockSomeMessages(lockId, now(), now().plus(lockDuration, SECONDS)); return eventLogRepository.findByLockedByAndLockedUntilGreaterThan(lockId, now()); } diff --git a/nakadi-producer/src/test/java/org/zalando/nakadiproducer/transmission/impl/EventTransmissionServiceTest.java b/nakadi-producer/src/test/java/org/zalando/nakadiproducer/transmission/impl/EventTransmissionServiceTest.java index b51cd7c3..fecfbf67 100644 --- a/nakadi-producer/src/test/java/org/zalando/nakadiproducer/transmission/impl/EventTransmissionServiceTest.java +++ b/nakadi-producer/src/test/java/org/zalando/nakadiproducer/transmission/impl/EventTransmissionServiceTest.java @@ -52,7 +52,7 @@ public void setUp() { repo = mock(EventLogRepository.class); publishingClient = spy(new MockNakadiPublishingClient()); mapper = spy(new ObjectMapper()); - service = new EventTransmissionService(repo, publishingClient, mapper, 0, 600, 60); + service = new EventTransmissionService(repo, publishingClient, mapper, 600, 60); } @Test From ec94e6b9545143472b71163a11171cd4fe580dc8 Mon Sep 17 00:00:00 2001 From: Florian Brons Date: Wed, 16 Nov 2022 11:26:11 +0100 Subject: [PATCH 3/4] #112 Do not lock all available events --- README.md | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 16308b63..3a2a4d6d 100644 --- a/README.md +++ b/README.md @@ -335,8 +335,9 @@ until the lock expires. The default is currently 600 seconds but might change in buffer is included. During the last x seconds before the expiration of the lock the events are not considered for transmission. The default is currently 60 seconds but might change in future releases. -* **lock-size**: Defines the maximum amount of events which are loaded into memory and published in one run - (in one submission per event type). By default, all events are loaded into memory. +* **lock-size**: Defines the maximum amount of events which are loaded into memory and published in one run +(in one submission per event type). By default, all events are loaded into memory. In future releases, this +property will become mandatory. ```yaml nakadi-producer: From 46982da69a18d0eda29c4fe684ab5c726a489126 Mon Sep 17 00:00:00 2001 From: Florian Brons Date: Wed, 16 Nov 2022 11:44:57 +0100 Subject: [PATCH 4/4] #112 Do not lock all available events --- nakadi-producer-loadtest/pom.xml | 2 +- nakadi-producer-spring-boot-starter/pom.xml | 2 +- nakadi-producer-starter-spring-boot-2-test/pom.xml | 4 ++-- nakadi-producer/pom.xml | 2 +- pom.xml | 2 +- 5 files changed, 6 insertions(+), 6 deletions(-) diff --git a/nakadi-producer-loadtest/pom.xml b/nakadi-producer-loadtest/pom.xml index a33fc775..d565455d 100644 --- a/nakadi-producer-loadtest/pom.xml +++ b/nakadi-producer-loadtest/pom.xml @@ -11,7 +11,7 @@ org.zalando nakadi-producer-reactor - 20.3.2 + 20.4.0 diff --git a/nakadi-producer-spring-boot-starter/pom.xml b/nakadi-producer-spring-boot-starter/pom.xml index a79813f0..2922cebf 100644 --- a/nakadi-producer-spring-boot-starter/pom.xml +++ b/nakadi-producer-spring-boot-starter/pom.xml @@ -10,7 +10,7 @@ org.zalando nakadi-producer-reactor - 20.3.2 + 20.4.0 nakadi-producer-spring-boot-starter diff --git a/nakadi-producer-starter-spring-boot-2-test/pom.xml b/nakadi-producer-starter-spring-boot-2-test/pom.xml index 2d434a14..8274d3ad 100644 --- a/nakadi-producer-starter-spring-boot-2-test/pom.xml +++ b/nakadi-producer-starter-spring-boot-2-test/pom.xml @@ -10,7 +10,7 @@ org.zalando nakadi-producer-reactor - 20.3.1 + 20.4.0 @@ -92,4 +92,4 @@ - \ No newline at end of file + diff --git a/nakadi-producer/pom.xml b/nakadi-producer/pom.xml index fbe6acfc..c054c322 100644 --- a/nakadi-producer/pom.xml +++ b/nakadi-producer/pom.xml @@ -10,7 +10,7 @@ org.zalando nakadi-producer-reactor - 20.3.2 + 20.4.0 nakadi-producer diff --git a/pom.xml b/pom.xml index 3b55880d..38af18ee 100644 --- a/pom.xml +++ b/pom.xml @@ -15,7 +15,7 @@ nakadi-producer-reactor org.zalando - 20.3.2 + 20.4.0 pom Nakadi Event Producer Reactor