Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fixes a race condition in tests

Browse Source 
https://build.spring.io/browse/INTEXT-KAFKA-JOB1-39
This commit is contained in:
Marius Bogoevici
2015-01-16 13:59:39 -05:00
committed by Artem Bilan
parent 6a6d119c4e
commit 2cae7420a3
3 changed files with 17 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterTests.java
View File

@@ -89,15 +89,16 @@ public class KafkaMessageDrivenChannelAdapterTests extends AbstractMessageListen
kafkaMessageDrivenChannelAdapter.setOutputChannel(new MessageChannel() {
@Override
public boolean send(Message<?> message) {
latch.countDown();
return receivedData.put(
(Integer)message.getHeaders().get(KafkaHeaders.PARTITION_ID),
boolean addedSuccessfully = receivedData.put(
(Integer) message.getHeaders().get(KafkaHeaders.PARTITION_ID),
new KeyedMessageWithOffset(
(String)message.getHeaders().get(KafkaHeaders.MESSAGE_KEY),
(String)message.getPayload(),
(Long)message.getHeaders().get(KafkaHeaders.OFFSET),
(String) message.getHeaders().get(KafkaHeaders.MESSAGE_KEY),
(String) message.getPayload(),
(Long) message.getHeaders().get(KafkaHeaders.OFFSET),
Thread.currentThread().getName(),
(Integer)message.getHeaders().get(KafkaHeaders.PARTITION_ID)));
(Integer) message.getHeaders().get(KafkaHeaders.PARTITION_ID)));
latch.countDown();
return addedSuccessfully;
}

5
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterWithSpecialOffsetTests.java
View File

@@ -110,8 +110,7 @@ public class KafkaMessageDrivenChannelAdapterWithSpecialOffsetTests extends Abst
kafkaMessageDrivenChannelAdapter.setOutputChannel(new MessageChannel() {
@Override
public boolean send(Message<?> message) {
latch.countDown();
return receivedData.put(
boolean addedSuccessfully = receivedData.put(
(Integer) message.getHeaders().get(KafkaHeaders.PARTITION_ID),
new KeyedMessageWithOffset(
(String) message.getHeaders().get(KafkaHeaders.MESSAGE_KEY),
@@ -119,6 +118,8 @@ public class KafkaMessageDrivenChannelAdapterWithSpecialOffsetTests extends Abst
(Long) message.getHeaders().get(KafkaHeaders.OFFSET),
Thread.currentThread().getName(),
(Integer) message.getHeaders().get(KafkaHeaders.PARTITION_ID)));
latch.countDown();
return addedSuccessfully;
}

10
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/listener/KafkaMessageDrivenChannelAdapterWithWrongOffsetTests.java
View File

@@ -113,8 +113,7 @@ public class KafkaMessageDrivenChannelAdapterWithWrongOffsetTests extends Abstra
kafkaMessageDrivenChannelAdapter.setOutputChannel(new MessageChannel() {
@Override
public boolean send(Message<?> message) {
latch.countDown();
return receivedData.put(
boolean addedSuccessfully = receivedData.put(
(Integer) message.getHeaders().get(KafkaHeaders.PARTITION_ID),
new KeyedMessageWithOffset(
(String) message.getHeaders().get(KafkaHeaders.MESSAGE_KEY),
@@ -122,6 +121,8 @@ public class KafkaMessageDrivenChannelAdapterWithWrongOffsetTests extends Abstra
(Long) message.getHeaders().get(KafkaHeaders.OFFSET),
Thread.currentThread().getName(),
(Integer) message.getHeaders().get(KafkaHeaders.PARTITION_ID)));
latch.countDown();
return addedSuccessfully;
}
@@ -254,8 +255,7 @@ public class KafkaMessageDrivenChannelAdapterWithWrongOffsetTests extends Abstra
kafkaMessageDrivenChannelAdapter.setOutputChannel(new MessageChannel() {
@Override
public boolean send(Message<?> message) {
latch.countDown();
return receivedData.put(
boolean addedSuccessfully = receivedData.put(
(Integer) message.getHeaders().get(KafkaHeaders.PARTITION_ID),
new KeyedMessageWithOffset(
(String) message.getHeaders().get(KafkaHeaders.MESSAGE_KEY),
@@ -263,6 +263,8 @@ public class KafkaMessageDrivenChannelAdapterWithWrongOffsetTests extends Abstra
(Long) message.getHeaders().get(KafkaHeaders.OFFSET),
Thread.currentThread().getName(),
(Integer) message.getHeaders().get(KafkaHeaders.PARTITION_ID)));
latch.countDown();
return addedSuccessfully;
}
@Override