You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2016/07/01 06:56:31 UTC
[1/6] camel git commit: Changed KafkaProducer to return
RecordMetaData in Exchange.
Repository: camel
Updated Branches:
refs/heads/master fbac9200a -> 0409cf089
Changed KafkaProducer to return RecordMetaData in Exchange.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/8919d65a
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/8919d65a
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/8919d65a
Branch: refs/heads/master
Commit: 8919d65a48f03ede1068547e74220a4121231cb4
Parents: fbac920
Author: Leo Prince <le...@target.com>
Authored: Tue Jun 28 10:03:32 2016 -0500
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri Jul 1 08:45:18 2016 +0200
----------------------------------------------------------------------
.../apache/camel/component/kafka/KafkaConstants.java | 1 +
.../apache/camel/component/kafka/KafkaProducer.java | 15 ++++++++++++---
.../camel/component/kafka/KafkaProducerTest.java | 14 ++++++++++++++
3 files changed, 27 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/8919d65a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConstants.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConstants.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConstants.java
index db99a09..1ae0759 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConstants.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConstants.java
@@ -30,6 +30,7 @@ public final class KafkaConstants {
public static final String KAFKA_DEFAULT_DESERIALIZER = "org.apache.kafka.common.serialization.StringDeserializer";
public static final String KAFKA_DEFAULT_PARTITIONER = "org.apache.kafka.clients.producer.internals.DefaultPartitioner";
public static final String PARTITIONER_RANGE_ASSIGNOR = "org.apache.kafka.clients.consumer.RangeAssignor";
+ public static final String KAFKA_RECORDMETA = "kafka.RECORDMETA";
private KafkaConstants() {
// Utility class
http://git-wip-us.apache.org/repos/asf/camel/blob/8919d65a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
index af41a24..da227c7 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
@@ -16,11 +16,13 @@
*/
package org.apache.camel.component.kafka;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Properties;
+import java.util.Vector;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
@@ -174,13 +176,16 @@ public class KafkaProducer extends DefaultAsyncProducer {
// Camel calls this method if the endpoint isSynchronous(), as the KafkaEndpoint creates a SynchronousDelegateProducer for it
public void process(Exchange exchange) throws Exception {
Iterator<ProducerRecord> c = createRecorder(exchange);
- List<Future<ProducerRecord>> futures = new LinkedList<Future<ProducerRecord>>();
+ List<Future<RecordMetadata>> futures = new LinkedList<Future<RecordMetadata>>();
+ List<RecordMetadata> recordMetadatas = new ArrayList<RecordMetadata>();
+ exchange.getOut().setHeader(KafkaConstants.KAFKA_RECORDMETA, recordMetadatas);
+
while (c.hasNext()) {
futures.add(kafkaProducer.send(c.next()));
}
- for (Future<ProducerRecord> f : futures) {
+ for (Future<RecordMetadata> f : futures) {
//wait for them all to be sent
- f.get();
+ recordMetadatas.add(f.get());
}
}
@@ -207,10 +212,12 @@ public class KafkaProducer extends DefaultAsyncProducer {
private final Exchange exchange;
private final AsyncCallback callback;
private final AtomicInteger count = new AtomicInteger(1);
+ private final List<RecordMetadata> recordMetadatas = new Vector<>();
KafkaProducerCallBack(Exchange exchange, AsyncCallback callback) {
this.exchange = exchange;
this.callback = callback;
+ exchange.getOut().setHeader(KafkaConstants.KAFKA_RECORDMETA, recordMetadatas);
}
void increment() {
@@ -230,6 +237,8 @@ public class KafkaProducer extends DefaultAsyncProducer {
if (e != null) {
exchange.setException(e);
}
+ recordMetadatas.add(recordMetadata);
+
if (count.decrementAndGet() == 0) {
// use worker pool to continue routing the exchange
// as this thread is from Kafka Callback and should not be used by Camel routing
http://git-wip-us.apache.org/repos/asf/camel/blob/8919d65a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
index dcd3365..1a29c4d 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
@@ -43,6 +43,7 @@ public class KafkaProducerTest {
private Exchange exchange = Mockito.mock(Exchange.class);
private Message in = new DefaultMessage();
+ private Message out = new DefaultMessage();
private AsyncCallback callback = Mockito.mock(AsyncCallback.class);
@SuppressWarnings({"unchecked"})
@@ -72,6 +73,8 @@ public class KafkaProducerTest {
public void processSendsMessage() throws Exception {
endpoint.setTopic("sometopic");
Mockito.when(exchange.getIn()).thenReturn(in);
+ Mockito.when(exchange.getOut()).thenReturn(out);
+
in.setHeader(KafkaConstants.PARTITION_KEY, "4");
producer.process(exchange);
@@ -95,6 +98,7 @@ public class KafkaProducerTest {
public void processAsyncSendsMessage() throws Exception {
endpoint.setTopic("sometopic");
Mockito.when(exchange.getIn()).thenReturn(in);
+ Mockito.when(exchange.getOut()).thenReturn(out);
in.setHeader(KafkaConstants.PARTITION_KEY, "4");
@@ -108,6 +112,7 @@ public class KafkaProducerTest {
endpoint.setTopic("sometopic");
Mockito.when(exchange.getIn()).thenReturn(in);
+ Mockito.when(exchange.getOut()).thenReturn(out);
// setup the exception here
org.apache.kafka.clients.producer.KafkaProducer kp = producer.getKafkaProducer();
@@ -127,6 +132,7 @@ public class KafkaProducerTest {
endpoint.setTopic(null);
Mockito.when(exchange.getIn()).thenReturn(in);
in.setHeader(KafkaConstants.TOPIC, "anotherTopic");
+ Mockito.when(exchange.getOut()).thenReturn(out);
producer.process(exchange);
@@ -137,6 +143,8 @@ public class KafkaProducerTest {
public void processSendsMessageWithTopicHeaderAndEndPoint() throws Exception {
endpoint.setTopic("sometopic");
Mockito.when(exchange.getIn()).thenReturn(in);
+ Mockito.when(exchange.getOut()).thenReturn(out);
+
in.setHeader(KafkaConstants.PARTITION_KEY, "4");
in.setHeader(KafkaConstants.TOPIC, "anotherTopic");
in.setHeader(KafkaConstants.KEY, "someKey");
@@ -158,6 +166,8 @@ public class KafkaProducerTest {
public void processDoesNotRequirePartitionHeader() throws Exception {
endpoint.setTopic("sometopic");
Mockito.when(exchange.getIn()).thenReturn(in);
+ Mockito.when(exchange.getOut()).thenReturn(out);
+
producer.process(exchange);
}
@@ -165,6 +175,7 @@ public class KafkaProducerTest {
public void processSendsMesssageWithPartitionKeyHeader() throws Exception {
endpoint.setTopic("someTopic");
Mockito.when(exchange.getIn()).thenReturn(in);
+ Mockito.when(exchange.getOut()).thenReturn(out);
in.setHeader(KafkaConstants.PARTITION_KEY, "4");
in.setHeader(KafkaConstants.KEY, "someKey");
producer.process(exchange);
@@ -175,6 +186,7 @@ public class KafkaProducerTest {
public void processSendsMesssageWithMessageKeyHeader() throws Exception {
endpoint.setTopic("someTopic");
Mockito.when(exchange.getIn()).thenReturn(in);
+ Mockito.when(exchange.getOut()).thenReturn(out);
in.setHeader(KafkaConstants.KEY, "someKey");
producer.process(exchange);
@@ -187,6 +199,7 @@ public class KafkaProducerTest {
endpoint.setTopic("someTopic");
endpoint.setBridgeEndpoint(true);
Mockito.when(exchange.getIn()).thenReturn(in);
+ Mockito.when(exchange.getOut()).thenReturn(out);
in.setHeader(KafkaConstants.TOPIC, "anotherTopic");
in.setHeader(KafkaConstants.KEY, "someKey");
in.setHeader(KafkaConstants.PARTITION_KEY, "4");
@@ -199,6 +212,7 @@ public class KafkaProducerTest {
public void processSendsMesssageWithMessageTopicName() throws Exception {
endpoint.setTopic("someTopic");
Mockito.when(exchange.getIn()).thenReturn(in);
+ Mockito.when(exchange.getOut()).thenReturn(out);
producer.process(exchange);
[2/6] camel git commit: Added Unit Tests.
Posted by da...@apache.org.
Added Unit Tests.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/a48c022c
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/a48c022c
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/a48c022c
Branch: refs/heads/master
Commit: a48c022c3d8424d65a6d23b3edf8854f0f38eaca
Parents: 8919d65
Author: Leo Prince <le...@target.com>
Authored: Tue Jun 28 13:04:23 2016 -0500
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri Jul 1 08:45:24 2016 +0200
----------------------------------------------------------------------
.../component/kafka/KafkaProducerFullTest.java | 54 +++++++++++++++++---
.../component/kafka/KafkaProducerTest.java | 48 ++++++++++++++++-
2 files changed, 94 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/a48c022c/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerFullTest.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerFullTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerFullTest.java
index 30f2b13..f1189d5 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerFullTest.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerFullTest.java
@@ -28,13 +28,16 @@ import java.util.concurrent.TimeUnit;
import org.apache.camel.Endpoint;
import org.apache.camel.EndpointInject;
+import org.apache.camel.Exchange;
import org.apache.camel.Produce;
import org.apache.camel.ProducerTemplate;
import org.apache.camel.RoutesBuilder;
import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.RecordMetadata;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
@@ -58,7 +61,8 @@ public class KafkaProducerFullTest extends BaseEmbeddedKafkaTest {
@EndpointInject(uri = "kafka:localhost:{{karfkaPort}}?topic=" + TOPIC_STRINGS
+ "&requestRequiredAcks=-1")
private Endpoint toStrings;
-
+ @EndpointInject(uri = "mock:kafkaAck")
+ private MockEndpoint mockEndpoint;
@EndpointInject(uri = "kafka:localhost:{{karfkaPort}}?topic=" + TOPIC_BYTES + "&requestRequiredAcks=-1"
+ "&serializerClass=org.apache.kafka.common.serialization.ByteArraySerializer&"
+ "keySerializerClass=org.apache.kafka.common.serialization.ByteArraySerializer")
@@ -103,9 +107,9 @@ public class KafkaProducerFullTest extends BaseEmbeddedKafkaTest {
return new RouteBuilder() {
@Override
public void configure() throws Exception {
- from("direct:startStrings").to(toStrings);
+ from("direct:startStrings").to(toStrings).to(mockEndpoint);
- from("direct:startBytes").to(toBytes);
+ from("direct:startBytes").to(toBytes).to(mockEndpoint);
}
};
}
@@ -125,9 +129,20 @@ public class KafkaProducerFullTest extends BaseEmbeddedKafkaTest {
boolean allMessagesReceived = messagesLatch.await(200, TimeUnit.MILLISECONDS);
assertTrue("Not all messages were published to the kafka topics. Not received: " + messagesLatch.getCount(), allMessagesReceived);
+
+ List<Exchange> exchangeList= mockEndpoint.getExchanges();
+ assertEquals("Fifteen Exchanges are expected",exchangeList.size(),15);
+ for (Exchange exchange : exchangeList) {
+ List<RecordMetadata> recordMetaData1 = (List<RecordMetadata>)(exchange.getIn().getHeader(KafkaConstants.KAFKA_RECORDMETA));
+ assertEquals("One RecordMetadata is expected.",recordMetaData1.size(),1);
+ assertTrue("Offset is positive",recordMetaData1.get(0).offset() >= 0);
+ assertTrue("Topic Name start with 'test'",recordMetaData1.get(0).topic().startsWith("test"));
+
+ }
+
}
-
+
@Test
public void producedStringCollectionMessageIsReceivedByKafka() throws InterruptedException, IOException {
int messageInTopic = 10;
@@ -139,7 +154,7 @@ public class KafkaProducerFullTest extends BaseEmbeddedKafkaTest {
for (int x = 0; x < messageInTopic; x++) {
msgs.add("Message " + x);
}
-
+
sendMessagesInRoute(1, stringsTemplate, msgs, KafkaConstants.PARTITION_KEY, "1");
msgs = new ArrayList<String>();
for (int x = 0; x < messageInOtherTopic; x++) {
@@ -152,8 +167,24 @@ public class KafkaProducerFullTest extends BaseEmbeddedKafkaTest {
boolean allMessagesReceived = messagesLatch.await(200, TimeUnit.MILLISECONDS);
assertTrue("Not all messages were published to the kafka topics. Not received: " + messagesLatch.getCount(), allMessagesReceived);
+ List<Exchange> exchangeList= mockEndpoint.getExchanges();
+ assertEquals("Two Exchanges are expected",exchangeList.size(),2);
+ Exchange e1 = exchangeList.get(0);
+ List<RecordMetadata> recordMetaData1 = (List<RecordMetadata>)(e1.getIn().getHeader(KafkaConstants.KAFKA_RECORDMETA));
+ assertEquals("Ten RecordMetadata is expected.",recordMetaData1.size(),10);
+ for (RecordMetadata recordMeta : recordMetaData1) {
+ assertTrue("Offset is positive",recordMeta.offset() >= 0);
+ assertTrue("Topic Name start with 'test'",recordMeta.topic().startsWith("test"));
+ }
+ Exchange e2 = exchangeList.get(1);
+ List<RecordMetadata> recordMetaData2 = (List<RecordMetadata>)(e2.getIn().getHeader(KafkaConstants.KAFKA_RECORDMETA));
+ assertEquals("Five RecordMetadata is expected.",recordMetaData2.size(),5);
+ for (RecordMetadata recordMeta : recordMetaData2) {
+ assertTrue("Offset is positive",recordMeta.offset() >= 0);
+ assertTrue("Topic Name start with 'test'",recordMeta.topic().startsWith("test"));
+ }
}
-
+
@Test
public void producedBytesMessageIsReceivedByKafka() throws InterruptedException, IOException {
int messageInTopic = 10;
@@ -175,6 +206,17 @@ public class KafkaProducerFullTest extends BaseEmbeddedKafkaTest {
boolean allMessagesReceived = messagesLatch.await(200, TimeUnit.MILLISECONDS);
assertTrue("Not all messages were published to the kafka topics. Not received: " + messagesLatch.getCount(), allMessagesReceived);
+
+ List<Exchange> exchangeList= mockEndpoint.getExchanges();
+ assertEquals("Fifteen Exchanges are expected",exchangeList.size(),15);
+ for (Exchange exchange : exchangeList) {
+ List<RecordMetadata> recordMetaData1 = (List<RecordMetadata>)(exchange.getIn().getHeader(KafkaConstants.KAFKA_RECORDMETA));
+ assertEquals("One RecordMetadata is expected.",recordMetaData1.size(),1);
+ assertTrue("Offset is positive",recordMetaData1.get(0).offset() >= 0);
+ assertTrue("Topic Name start with 'test'",recordMetaData1.get(0).topic().startsWith("test"));
+
+ }
+
}
private void createKafkaMessageConsumer(KafkaConsumer<String, String> consumerConn,
http://git-wip-us.apache.org/repos/asf/camel/blob/a48c022c/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
index 1a29c4d..2dcd8f8 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
@@ -16,7 +16,9 @@
*/
package org.apache.camel.component.kafka;
+import java.util.List;
import java.util.Properties;
+import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import org.apache.camel.AsyncCallback;
@@ -35,6 +37,8 @@ import org.mockito.Matchers;
import org.mockito.Mockito;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
public class KafkaProducerTest {
@@ -60,6 +64,7 @@ public class KafkaProducerTest {
Mockito.when(kp.send(Mockito.any(ProducerRecord.class))).thenReturn(future);
producer.setKafkaProducer(kp);
+ producer.setWorkerPool(Executors.newFixedThreadPool(1));
}
@Test
@@ -79,6 +84,9 @@ public class KafkaProducerTest {
producer.process(exchange);
Mockito.verify(producer.getKafkaProducer()).send(Matchers.any(ProducerRecord.class));
+ assertRecordMetadataExists();
+
+
}
@Test(expected = Exception.class)
@@ -92,8 +100,11 @@ public class KafkaProducerTest {
in.setHeader(KafkaConstants.PARTITION_KEY, "4");
producer.process(exchange);
+
+ assertRecordMetadataExists();
}
+
@Test
public void processAsyncSendsMessage() throws Exception {
endpoint.setTopic("sometopic");
@@ -104,7 +115,12 @@ public class KafkaProducerTest {
producer.process(exchange, callback);
- Mockito.verify(producer.getKafkaProducer()).send(Matchers.any(ProducerRecord.class), Matchers.any(Callback.class));
+ ArgumentCaptor<Callback> callBackCaptor = ArgumentCaptor.forClass(Callback.class);
+ Mockito.verify(producer.getKafkaProducer()).send(Matchers.any(ProducerRecord.class), callBackCaptor.capture());
+ Callback kafkaCallback = callBackCaptor.getValue();
+ kafkaCallback.onCompletion(new RecordMetadata(null, 1, 1),null);
+ assertRecordMetadataExists();
+
}
@Test
@@ -122,9 +138,13 @@ public class KafkaProducerTest {
producer.process(exchange, callback);
- Mockito.verify(producer.getKafkaProducer()).send(Matchers.any(ProducerRecord.class), Matchers.any(Callback.class));
+ ArgumentCaptor<Callback> callBackCaptor = ArgumentCaptor.forClass(Callback.class);
+ Mockito.verify(producer.getKafkaProducer()).send(Matchers.any(ProducerRecord.class), callBackCaptor.capture());
Mockito.verify(exchange).setException(Matchers.isA(ApiException.class));
Mockito.verify(callback).done(Matchers.eq(true));
+ Callback kafkaCallback = callBackCaptor.getValue();
+ kafkaCallback.onCompletion(new RecordMetadata(null, 1, 1),null);
+ assertRecordMetadataExists();
}
@Test
@@ -137,6 +157,7 @@ public class KafkaProducerTest {
producer.process(exchange);
verifySendMessage("anotherTopic");
+ assertRecordMetadataExists();
}
@Test
@@ -152,6 +173,7 @@ public class KafkaProducerTest {
producer.process(exchange);
verifySendMessage("4", "anotherTopic", "someKey");
+ assertRecordMetadataExists();
}
@Test(expected = CamelException.class)
@@ -159,7 +181,10 @@ public class KafkaProducerTest {
endpoint.setTopic(null);
Mockito.when(exchange.getIn()).thenReturn(in);
in.setHeader(KafkaConstants.PARTITION_KEY, "4");
+
producer.process(exchange);
+
+ assertRecordMetadataExists();
}
@Test
@@ -169,6 +194,8 @@ public class KafkaProducerTest {
Mockito.when(exchange.getOut()).thenReturn(out);
producer.process(exchange);
+
+ assertRecordMetadataExists();
}
@Test
@@ -178,8 +205,12 @@ public class KafkaProducerTest {
Mockito.when(exchange.getOut()).thenReturn(out);
in.setHeader(KafkaConstants.PARTITION_KEY, "4");
in.setHeader(KafkaConstants.KEY, "someKey");
+
producer.process(exchange);
+
verifySendMessage("4", "someTopic", "someKey");
+ assertRecordMetadataExists();
+
}
@Test
@@ -192,6 +223,8 @@ public class KafkaProducerTest {
producer.process(exchange);
verifySendMessage("someTopic", "someKey");
+ assertRecordMetadataExists();
+
}
@Test
@@ -203,9 +236,12 @@ public class KafkaProducerTest {
in.setHeader(KafkaConstants.TOPIC, "anotherTopic");
in.setHeader(KafkaConstants.KEY, "someKey");
in.setHeader(KafkaConstants.PARTITION_KEY, "4");
+
producer.process(exchange);
verifySendMessage("4", "someTopic", "someKey");
+ assertRecordMetadataExists();
+
}
@Test // Message and Topic Name alone
@@ -217,6 +253,8 @@ public class KafkaProducerTest {
producer.process(exchange);
verifySendMessage("someTopic");
+ assertRecordMetadataExists();
+
}
@SuppressWarnings({"unchecked", "rawtypes"})
@@ -243,4 +281,10 @@ public class KafkaProducerTest {
assertEquals(topic, captor.getValue().topic());
}
+ private void assertRecordMetadataExists() {
+ List<RecordMetadata> recordMetaData1 = (List<RecordMetadata>)(out.getHeader(KafkaConstants.KAFKA_RECORDMETA));
+ assertTrue(recordMetaData1 != null);
+ assertEquals("Expected one recordMetaData",recordMetaData1.size(),1);
+ assertTrue(recordMetaData1.get(0) !=null);
+ }
}
[3/6] camel git commit: Fixed review comments for pull request 1052.
Posted by da...@apache.org.
Fixed review comments for pull request 1052.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/dc48fe7c
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/dc48fe7c
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/dc48fe7c
Branch: refs/heads/master
Commit: dc48fe7c2cdf01b7654e0faa937272b8a176969d
Parents: a48c022
Author: Leo Prince <le...@target.com>
Authored: Wed Jun 29 09:03:53 2016 -0500
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri Jul 1 08:45:31 2016 +0200
----------------------------------------------------------------------
.../org/apache/camel/component/kafka/KafkaConstants.java | 2 +-
.../java/org/apache/camel/component/kafka/KafkaProducer.java | 8 ++++----
2 files changed, 5 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/dc48fe7c/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConstants.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConstants.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConstants.java
index 1ae0759..2bd4c4d 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConstants.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConstants.java
@@ -30,7 +30,7 @@ public final class KafkaConstants {
public static final String KAFKA_DEFAULT_DESERIALIZER = "org.apache.kafka.common.serialization.StringDeserializer";
public static final String KAFKA_DEFAULT_PARTITIONER = "org.apache.kafka.clients.producer.internals.DefaultPartitioner";
public static final String PARTITIONER_RANGE_ASSIGNOR = "org.apache.kafka.clients.consumer.RangeAssignor";
- public static final String KAFKA_RECORDMETA = "kafka.RECORDMETA";
+ public static final String KAFKA_RECORDMETA = "org.apache.kafka.clients.producer.RecordMetadata";
private KafkaConstants() {
// Utility class
http://git-wip-us.apache.org/repos/asf/camel/blob/dc48fe7c/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
index da227c7..bba8d82 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
@@ -38,7 +38,7 @@ import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
public class KafkaProducer extends DefaultAsyncProducer {
-
+
private org.apache.kafka.clients.producer.KafkaProducer kafkaProducer;
private final KafkaEndpoint endpoint;
private ExecutorService workerPool;
@@ -48,7 +48,7 @@ public class KafkaProducer extends DefaultAsyncProducer {
super(endpoint);
this.endpoint = endpoint;
}
-
+
Properties getProps() {
Properties props = endpoint.getConfiguration().createProducerProperties();
endpoint.updateClassProperties(props);
@@ -212,7 +212,7 @@ public class KafkaProducer extends DefaultAsyncProducer {
private final Exchange exchange;
private final AsyncCallback callback;
private final AtomicInteger count = new AtomicInteger(1);
- private final List<RecordMetadata> recordMetadatas = new Vector<>();
+ private final List<RecordMetadata> recordMetadatas = new ArrayList<>();
KafkaProducerCallBack(Exchange exchange, AsyncCallback callback) {
this.exchange = exchange;
@@ -231,7 +231,7 @@ public class KafkaProducer extends DefaultAsyncProducer {
}
return false;
}
-
+
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
if (e != null) {
[6/6] camel git commit: CAMEL-10020: Fixed CS
Posted by da...@apache.org.
CAMEL-10020: Fixed CS
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/0409cf08
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/0409cf08
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/0409cf08
Branch: refs/heads/master
Commit: 0409cf089d2143a61bc1d478077d51ba45f0737d
Parents: 466e03c
Author: Claus Ibsen <da...@apache.org>
Authored: Fri Jul 1 08:55:25 2016 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri Jul 1 08:55:25 2016 +0200
----------------------------------------------------------------------
.../component/kafka/KafkaProducerFullTest.java | 50 +++++++++-----------
.../component/kafka/KafkaProducerTest.java | 8 ++--
2 files changed, 26 insertions(+), 32 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/0409cf08/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerFullTest.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerFullTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerFullTest.java
index f1189d5..88f0177 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerFullTest.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerFullTest.java
@@ -31,7 +31,6 @@ import org.apache.camel.EndpointInject;
import org.apache.camel.Exchange;
import org.apache.camel.Produce;
import org.apache.camel.ProducerTemplate;
-import org.apache.camel.RoutesBuilder;
import org.apache.camel.builder.RouteBuilder;
import org.apache.camel.component.mock.MockEndpoint;
import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -130,19 +129,16 @@ public class KafkaProducerFullTest extends BaseEmbeddedKafkaTest {
assertTrue("Not all messages were published to the kafka topics. Not received: " + messagesLatch.getCount(), allMessagesReceived);
- List<Exchange> exchangeList= mockEndpoint.getExchanges();
- assertEquals("Fifteen Exchanges are expected",exchangeList.size(),15);
+ List<Exchange> exchangeList = mockEndpoint.getExchanges();
+ assertEquals("Fifteen Exchanges are expected", exchangeList.size(), 15);
for (Exchange exchange : exchangeList) {
- List<RecordMetadata> recordMetaData1 = (List<RecordMetadata>)(exchange.getIn().getHeader(KafkaConstants.KAFKA_RECORDMETA));
- assertEquals("One RecordMetadata is expected.",recordMetaData1.size(),1);
- assertTrue("Offset is positive",recordMetaData1.get(0).offset() >= 0);
- assertTrue("Topic Name start with 'test'",recordMetaData1.get(0).topic().startsWith("test"));
-
+ List<RecordMetadata> recordMetaData1 = (List<RecordMetadata>) (exchange.getIn().getHeader(KafkaConstants.KAFKA_RECORDMETA));
+ assertEquals("One RecordMetadata is expected.", recordMetaData1.size(), 1);
+ assertTrue("Offset is positive", recordMetaData1.get(0).offset() >= 0);
+ assertTrue("Topic Name start with 'test'", recordMetaData1.get(0).topic().startsWith("test"));
}
-
}
-
@Test
public void producedStringCollectionMessageIsReceivedByKafka() throws InterruptedException, IOException {
int messageInTopic = 10;
@@ -167,21 +163,21 @@ public class KafkaProducerFullTest extends BaseEmbeddedKafkaTest {
boolean allMessagesReceived = messagesLatch.await(200, TimeUnit.MILLISECONDS);
assertTrue("Not all messages were published to the kafka topics. Not received: " + messagesLatch.getCount(), allMessagesReceived);
- List<Exchange> exchangeList= mockEndpoint.getExchanges();
- assertEquals("Two Exchanges are expected",exchangeList.size(),2);
+ List<Exchange> exchangeList = mockEndpoint.getExchanges();
+ assertEquals("Two Exchanges are expected", exchangeList.size(), 2);
Exchange e1 = exchangeList.get(0);
- List<RecordMetadata> recordMetaData1 = (List<RecordMetadata>)(e1.getIn().getHeader(KafkaConstants.KAFKA_RECORDMETA));
- assertEquals("Ten RecordMetadata is expected.",recordMetaData1.size(),10);
+ List<RecordMetadata> recordMetaData1 = (List<RecordMetadata>) (e1.getIn().getHeader(KafkaConstants.KAFKA_RECORDMETA));
+ assertEquals("Ten RecordMetadata is expected.", recordMetaData1.size(), 10);
for (RecordMetadata recordMeta : recordMetaData1) {
- assertTrue("Offset is positive",recordMeta.offset() >= 0);
- assertTrue("Topic Name start with 'test'",recordMeta.topic().startsWith("test"));
+ assertTrue("Offset is positive", recordMeta.offset() >= 0);
+ assertTrue("Topic Name start with 'test'", recordMeta.topic().startsWith("test"));
}
Exchange e2 = exchangeList.get(1);
- List<RecordMetadata> recordMetaData2 = (List<RecordMetadata>)(e2.getIn().getHeader(KafkaConstants.KAFKA_RECORDMETA));
- assertEquals("Five RecordMetadata is expected.",recordMetaData2.size(),5);
+ List<RecordMetadata> recordMetaData2 = (List<RecordMetadata>) (e2.getIn().getHeader(KafkaConstants.KAFKA_RECORDMETA));
+ assertEquals("Five RecordMetadata is expected.", recordMetaData2.size(), 5);
for (RecordMetadata recordMeta : recordMetaData2) {
- assertTrue("Offset is positive",recordMeta.offset() >= 0);
- assertTrue("Topic Name start with 'test'",recordMeta.topic().startsWith("test"));
+ assertTrue("Offset is positive", recordMeta.offset() >= 0);
+ assertTrue("Topic Name start with 'test'", recordMeta.topic().startsWith("test"));
}
}
@@ -207,16 +203,14 @@ public class KafkaProducerFullTest extends BaseEmbeddedKafkaTest {
assertTrue("Not all messages were published to the kafka topics. Not received: " + messagesLatch.getCount(), allMessagesReceived);
- List<Exchange> exchangeList= mockEndpoint.getExchanges();
- assertEquals("Fifteen Exchanges are expected",exchangeList.size(),15);
+ List<Exchange> exchangeList = mockEndpoint.getExchanges();
+ assertEquals("Fifteen Exchanges are expected", exchangeList.size(), 15);
for (Exchange exchange : exchangeList) {
- List<RecordMetadata> recordMetaData1 = (List<RecordMetadata>)(exchange.getIn().getHeader(KafkaConstants.KAFKA_RECORDMETA));
- assertEquals("One RecordMetadata is expected.",recordMetaData1.size(),1);
- assertTrue("Offset is positive",recordMetaData1.get(0).offset() >= 0);
- assertTrue("Topic Name start with 'test'",recordMetaData1.get(0).topic().startsWith("test"));
-
+ List<RecordMetadata> recordMetaData1 = (List<RecordMetadata>) (exchange.getIn().getHeader(KafkaConstants.KAFKA_RECORDMETA));
+ assertEquals("One RecordMetadata is expected.", recordMetaData1.size(), 1);
+ assertTrue("Offset is positive", recordMetaData1.get(0).offset() >= 0);
+ assertTrue("Topic Name start with 'test'", recordMetaData1.get(0).topic().startsWith("test"));
}
-
}
private void createKafkaMessageConsumer(KafkaConsumer<String, String> consumerConn,
http://git-wip-us.apache.org/repos/asf/camel/blob/0409cf08/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
index c10266e..449349f 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
@@ -114,7 +114,7 @@ public class KafkaProducerTest {
ArgumentCaptor<Callback> callBackCaptor = ArgumentCaptor.forClass(Callback.class);
Mockito.verify(producer.getKafkaProducer()).send(Matchers.any(ProducerRecord.class), callBackCaptor.capture());
Callback kafkaCallback = callBackCaptor.getValue();
- kafkaCallback.onCompletion(new RecordMetadata(null, 1, 1),null);
+ kafkaCallback.onCompletion(new RecordMetadata(null, 1, 1), null);
assertRecordMetadataExists();
}
@@ -138,7 +138,7 @@ public class KafkaProducerTest {
Mockito.verify(exchange).setException(Matchers.isA(ApiException.class));
Mockito.verify(callback).done(Matchers.eq(true));
Callback kafkaCallback = callBackCaptor.getValue();
- kafkaCallback.onCompletion(new RecordMetadata(null, 1, 1),null);
+ kafkaCallback.onCompletion(new RecordMetadata(null, 1, 1), null);
assertRecordMetadataExists();
}
@@ -273,9 +273,9 @@ public class KafkaProducerTest {
}
private void assertRecordMetadataExists() {
- List<RecordMetadata> recordMetaData1 = (List<RecordMetadata>)(in.getHeader(KafkaConstants.KAFKA_RECORDMETA));
+ List<RecordMetadata> recordMetaData1 = (List<RecordMetadata>) in.getHeader(KafkaConstants.KAFKA_RECORDMETA);
assertTrue(recordMetaData1 != null);
- assertEquals("Expected one recordMetaData",recordMetaData1.size(),1);
+ assertEquals("Expected one recordMetaData", recordMetaData1.size(), 1);
assertTrue(recordMetaData1.get(0) != null);
}
}
[4/6] camel git commit: Removed unused imports.
Posted by da...@apache.org.
Removed unused imports.
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/a81e4d7e
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/a81e4d7e
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/a81e4d7e
Branch: refs/heads/master
Commit: a81e4d7e35254e6dd39503f341bc240af99f291e
Parents: dc48fe7
Author: Leo Prince <le...@target.com>
Authored: Wed Jun 29 09:06:18 2016 -0500
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri Jul 1 08:45:38 2016 +0200
----------------------------------------------------------------------
.../main/java/org/apache/camel/component/kafka/KafkaProducer.java | 1 -
1 file changed, 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/a81e4d7e/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
index bba8d82..3b838e9 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
@@ -22,7 +22,6 @@ import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Properties;
-import java.util.Vector;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
[5/6] camel git commit: CAMEL-10020: Added option on endpoint
Posted by da...@apache.org.
CAMEL-10020: Added option on endpoint
Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/466e03c4
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/466e03c4
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/466e03c4
Branch: refs/heads/master
Commit: 466e03c4ec105693d862428e9a471e16857a0d40
Parents: a81e4d7
Author: Claus Ibsen <da...@apache.org>
Authored: Fri Jul 1 08:42:43 2016 +0200
Committer: Claus Ibsen <da...@apache.org>
Committed: Fri Jul 1 08:52:28 2016 +0200
----------------------------------------------------------------------
components/camel-kafka/src/main/docs/kafka.adoc | 5 ++++-
.../component/kafka/KafkaConfiguration.java | 18 ++++++++++++++++++
.../camel/component/kafka/KafkaProducer.java | 19 +++++++++++++++++--
.../camel/component/kafka/KafkaProducerTest.java | 13 ++-----------
4 files changed, 41 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/camel/blob/466e03c4/components/camel-kafka/src/main/docs/kafka.adoc
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/main/docs/kafka.adoc b/components/camel-kafka/src/main/docs/kafka.adoc
index 2e90613..8a07a13 100644
--- a/components/camel-kafka/src/main/docs/kafka.adoc
+++ b/components/camel-kafka/src/main/docs/kafka.adoc
@@ -85,8 +85,9 @@ The Kafka component supports 1 options which are listed below.
+
// endpoint options: START
-The Kafka component supports 74 endpoint options which are listed below:
+The Kafka component supports 75 endpoint options which are listed below:
{% raw %}
[width="100%",cols="2s,1,1m,1m,5",options="header"]
@@ -139,6 +140,7 @@ The Kafka component supports 74 endpoint options which are listed below:
| queueBufferingMaxMessages | producer | 10000 | Integer | The maximum number of unsent messages that can be queued up the producer when using async mode before either the producer must be blocked or data must be dropped.
| receiveBufferBytes | producer | 32768 | Integer | The size of the TCP receive buffer (SO_RCVBUF) to use when reading data.
| reconnectBackoffMs | producer | 50 | Integer | The amount of time to wait before attempting to reconnect to a given host. This avoids repeatedly connecting to a host in a tight loop. This backoff applies to all requests sent by the consumer to the broker.
+| recordMetadata | producer | true | boolean | Whether the producer should store the RecordMetadata results from sending to Kafka. The results are stored in a List containing the RecordMetadata metadata's. The list is stored on a header with the key link KafkaConstantsKAFKA_RECORDMETA
| requestRequiredAcks | producer | 1 | String | The number of acknowledgments the producer requires the leader to have received before considering a request complete. This controls the durability of records that are sent. The following settings are common: acks=0 If set to zero then the producer will not wait for any acknowledgment from the server at all. The record will be immediately added to the socket buffer and considered sent. No guarantee can be made that the server has received the record in this case and the retries configuration will not take effect (as the client won't generally know of any failures). The offset given back for each record will always be set to -1. acks=1 This will mean the leader will write the record to its local log but will respond without awaiting full acknowledgement from all followers. In this case should the leader fail immediately after acknowledging the record but before the followers have replicated it then the record will be lost. acks=all This
means the leader will wait for the full set of in-sync replicas to acknowledge the record. This guarantees that the record will not be lost as long as at least one in-sync replica remains alive. This is the strongest available guarantee.
| requestTimeoutMs | producer | 30000 | Integer | The amount of time the broker will wait trying to meet the request.required.acks requirement before sending back an error to the client.
| retries | producer | 0 | Integer | Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error. Note that this retry is no different than if the client resent the record upon receiving the error. Allowing retries will potentially change the ordering of records because if two records are sent to a single partition and the first fails and is retried but the second succeeds then the second record may appear first.
@@ -177,6 +179,7 @@ The Kafka component supports 74 endpoint options which are listed below:
+
For more information about Producer/Consumer configuration:
http://kafka.apache.org/documentation.html#newconsumerconfigs[http://kafka.apache.org/documentation.html#newconsumerconfigs]
http://git-wip-us.apache.org/repos/asf/camel/blob/466e03c4/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
index c066520..7a7f9d4 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaConfiguration.java
@@ -16,6 +16,7 @@
*/
package org.apache.camel.component.kafka;
+import java.util.List;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
@@ -27,6 +28,7 @@ import org.apache.camel.spi.UriPath;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
@@ -237,6 +239,8 @@ public class KafkaConfiguration {
//ssl.trustmanager.algorithm
@UriParam(label = "producer", defaultValue = "PKIX")
private String sslTrustmanagerAlgorithm = "PKIX";
+ @UriParam(label = "producer", defaultValue = "true")
+ private boolean recordMetadata = true;
public KafkaConfiguration() {
}
@@ -1202,4 +1206,18 @@ public class KafkaConfiguration {
public void setWorkerPoolMaxSize(Integer workerPoolMaxSize) {
this.workerPoolMaxSize = workerPoolMaxSize;
}
+
+ public boolean isRecordMetadata() {
+ return recordMetadata;
+ }
+
+ /**
+ * Whether the producer should store the {@link RecordMetadata} results from sending to Kafka.
+ *
+ * The results are stored in a {@link List} containing the {@link RecordMetadata} metadata's.
+ * The list is stored on a header with the key {@link KafkaConstants#KAFKA_RECORDMETA}
+ */
+ public void setRecordMetadata(boolean recordMetadata) {
+ this.recordMetadata = recordMetadata;
+ }
}
http://git-wip-us.apache.org/repos/asf/camel/blob/466e03c4/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
index 3b838e9..322c22f 100644
--- a/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
+++ b/components/camel-kafka/src/main/java/org/apache/camel/component/kafka/KafkaProducer.java
@@ -177,7 +177,14 @@ public class KafkaProducer extends DefaultAsyncProducer {
Iterator<ProducerRecord> c = createRecorder(exchange);
List<Future<RecordMetadata>> futures = new LinkedList<Future<RecordMetadata>>();
List<RecordMetadata> recordMetadatas = new ArrayList<RecordMetadata>();
- exchange.getOut().setHeader(KafkaConstants.KAFKA_RECORDMETA, recordMetadatas);
+
+ if (endpoint.getConfiguration().isRecordMetadata()) {
+ if (exchange.hasOut()) {
+ exchange.getOut().setHeader(KafkaConstants.KAFKA_RECORDMETA, recordMetadatas);
+ } else {
+ exchange.getIn().setHeader(KafkaConstants.KAFKA_RECORDMETA, recordMetadatas);
+ }
+ }
while (c.hasNext()) {
futures.add(kafkaProducer.send(c.next()));
@@ -216,12 +223,19 @@ public class KafkaProducer extends DefaultAsyncProducer {
KafkaProducerCallBack(Exchange exchange, AsyncCallback callback) {
this.exchange = exchange;
this.callback = callback;
- exchange.getOut().setHeader(KafkaConstants.KAFKA_RECORDMETA, recordMetadatas);
+ if (endpoint.getConfiguration().isRecordMetadata()) {
+ if (exchange.hasOut()) {
+ exchange.getOut().setHeader(KafkaConstants.KAFKA_RECORDMETA, recordMetadatas);
+ } else {
+ exchange.getIn().setHeader(KafkaConstants.KAFKA_RECORDMETA, recordMetadatas);
+ }
+ }
}
void increment() {
count.incrementAndGet();
}
+
boolean allSent() {
if (count.decrementAndGet() == 0) {
//was able to get all the work done while queuing the requests
@@ -236,6 +250,7 @@ public class KafkaProducer extends DefaultAsyncProducer {
if (e != null) {
exchange.setException(e);
}
+
recordMetadatas.add(recordMetadata);
if (count.decrementAndGet() == 0) {
http://git-wip-us.apache.org/repos/asf/camel/blob/466e03c4/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
----------------------------------------------------------------------
diff --git a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
index 2dcd8f8..c10266e 100644
--- a/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
+++ b/components/camel-kafka/src/test/java/org/apache/camel/component/kafka/KafkaProducerTest.java
@@ -39,7 +39,6 @@ import org.mockito.Mockito;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
-
public class KafkaProducerTest {
private KafkaProducer producer;
@@ -85,8 +84,6 @@ public class KafkaProducerTest {
producer.process(exchange);
Mockito.verify(producer.getKafkaProducer()).send(Matchers.any(ProducerRecord.class));
assertRecordMetadataExists();
-
-
}
@Test(expected = Exception.class)
@@ -104,7 +101,6 @@ public class KafkaProducerTest {
assertRecordMetadataExists();
}
-
@Test
public void processAsyncSendsMessage() throws Exception {
endpoint.setTopic("sometopic");
@@ -125,7 +121,6 @@ public class KafkaProducerTest {
@Test
public void processAsyncSendsMessageWithException() throws Exception {
-
endpoint.setTopic("sometopic");
Mockito.when(exchange.getIn()).thenReturn(in);
Mockito.when(exchange.getOut()).thenReturn(out);
@@ -210,7 +205,6 @@ public class KafkaProducerTest {
verifySendMessage("4", "someTopic", "someKey");
assertRecordMetadataExists();
-
}
@Test
@@ -224,7 +218,6 @@ public class KafkaProducerTest {
verifySendMessage("someTopic", "someKey");
assertRecordMetadataExists();
-
}
@Test
@@ -241,7 +234,6 @@ public class KafkaProducerTest {
verifySendMessage("4", "someTopic", "someKey");
assertRecordMetadataExists();
-
}
@Test // Message and Topic Name alone
@@ -254,7 +246,6 @@ public class KafkaProducerTest {
verifySendMessage("someTopic");
assertRecordMetadataExists();
-
}
@SuppressWarnings({"unchecked", "rawtypes"})
@@ -282,9 +273,9 @@ public class KafkaProducerTest {
}
private void assertRecordMetadataExists() {
- List<RecordMetadata> recordMetaData1 = (List<RecordMetadata>)(out.getHeader(KafkaConstants.KAFKA_RECORDMETA));
+ List<RecordMetadata> recordMetaData1 = (List<RecordMetadata>)(in.getHeader(KafkaConstants.KAFKA_RECORDMETA));
assertTrue(recordMetaData1 != null);
assertEquals("Expected one recordMetaData",recordMetaData1.size(),1);
- assertTrue(recordMetaData1.get(0) !=null);
+ assertTrue(recordMetaData1.get(0) != null);
}
}