You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2018/01/31 17:13:49 UTC
[GitHub] merlimat closed pull request #1115: In Kafka consumer wrapper, group multiple messages when the app polls
merlimat closed pull request #1115: In Kafka consumer wrapper, group multiple messages when the app polls
URL: https://github.com/apache/incubator-pulsar/pull/1115
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/pulsar-client-kafka-compat/pulsar-client-kafka-tests/src/test/java/org/apache/pulsar/client/kafka/compat/tests/KafkaApiTest.java b/pulsar-client-kafka-compat/pulsar-client-kafka-tests/src/test/java/org/apache/pulsar/client/kafka/compat/tests/KafkaApiTest.java
index 44357ea8f..957f3aa02 100644
--- a/pulsar-client-kafka-compat/pulsar-client-kafka-tests/src/test/java/org/apache/pulsar/client/kafka/compat/tests/KafkaApiTest.java
+++ b/pulsar-client-kafka-compat/pulsar-client-kafka-tests/src/test/java/org/apache/pulsar/client/kafka/compat/tests/KafkaApiTest.java
@@ -24,6 +24,7 @@
import java.util.Arrays;
import java.util.List;
import java.util.Properties;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
@@ -91,17 +92,18 @@ public void testSimpleProducerConsumer() throws Exception {
producer.flush();
producer.close();
- for (int i = 0; i < 10; i++) {
- ConsumerRecords<Integer, String> records = consumer.poll(1000);
- assertEquals(records.count(), 1);
-
- int idx = i;
+ AtomicInteger received = new AtomicInteger();
+ while (received.get() < 10) {
+ ConsumerRecords<Integer, String> records = consumer.poll(100);
records.forEach(record -> {
- log.info("Received record: {}", record);
- assertEquals(record.key().intValue(), idx);
- assertEquals(record.value(), "hello-" + idx);
- assertEquals(record.offset(), offsets.get(idx).longValue());
+ assertEquals(record.key().intValue(), received.get());
+ assertEquals(record.value(), "hello-" + received.get());
+ assertEquals(record.offset(), offsets.get(received.get()).longValue());
+
+ received.incrementAndGet();
});
+
+ consumer.commitSync();
}
consumer.close();
diff --git a/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java b/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java
index 379bde4ce..d3dc6e431 100644
--- a/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java
+++ b/pulsar-client-kafka-compat/pulsar-client-kafka/src/main/java/org/apache/kafka/clients/consumer/PulsarKafkaConsumer.java
@@ -61,8 +61,6 @@
import org.apache.pulsar.client.util.FutureUtil;
import org.apache.pulsar.common.naming.DestinationName;
-import com.google.common.collect.Lists;
-
public class PulsarKafkaConsumer<K, V> implements Consumer<K, V>, MessageListener {
private static final long serialVersionUID = 1L;
@@ -146,6 +144,9 @@ private PulsarKafkaConsumer(ConsumerConfig config, Deserializer<K> keyDeserializ
Properties properties = new Properties();
config.originals().forEach((k, v) -> properties.put(k, v));
ClientConfiguration clientConf = PulsarKafkaConfig.getClientConfiguration(properties);
+ // Since this client instance is going to be used just for the consumers, we can enable Nagle to group
+ // all the acknowledgments sent to broker within a short time frame
+ clientConf.setUseTcpNoDelay(false);
try {
client = PulsarClient.create(serviceUrl, clientConf);
} catch (PulsarClientException e) {
@@ -270,6 +271,8 @@ public void unsubscribe() {
});
}
+ private static final int MAX_RECORDS_IN_SINGLE_POLL = 1000;
+
@SuppressWarnings("unchecked")
@Override
public ConsumerRecords<K, V> poll(long timeoutMillis) {
@@ -279,40 +282,49 @@ public void unsubscribe() {
return (ConsumerRecords<K, V>) ConsumerRecords.EMPTY;
}
- if (isAutoCommit) {
- // Commit the offset of previously dequeued messages
- commitAsync();
- }
+ Map<TopicPartition, List<ConsumerRecord<K, V>>> records = new HashMap<>();
- DestinationName dn = DestinationName.get(item.consumer.getTopic());
- String topic = dn.getPartitionedTopicName();
- int partition = dn.isPartitioned() ? dn.getPartitionIndex() : 0;
- Message msg = item.message;
- MessageIdImpl msgId = (MessageIdImpl) msg.getMessageId();
- long offset = MessageIdUtils.getOffset(msgId);
+ int numberOfRecords = 0;
- TopicPartition tp = new TopicPartition(topic, partition);
+ while (item != null && ++numberOfRecords < MAX_RECORDS_IN_SINGLE_POLL) {
+ DestinationName dn = DestinationName.get(item.consumer.getTopic());
+ String topic = dn.getPartitionedTopicName();
+ int partition = dn.isPartitioned() ? dn.getPartitionIndex() : 0;
+ Message msg = item.message;
+ MessageIdImpl msgId = (MessageIdImpl) msg.getMessageId();
+ long offset = MessageIdUtils.getOffset(msgId);
- K key = getKey(topic, msg);
- V value = valueDeserializer.deserialize(topic, msg.getData());
+ TopicPartition tp = new TopicPartition(topic, partition);
- TimestampType timestampType = TimestampType.LOG_APPEND_TIME;
- long timestamp = msg.getPublishTime();
+ K key = getKey(topic, msg);
+ V value = valueDeserializer.deserialize(topic, msg.getData());
- if (msg.getEventTime() > 0) {
- // If we have Event time, use that in preference
- timestamp = msg.getEventTime();
- timestampType = TimestampType.CREATE_TIME;
- }
+ TimestampType timestampType = TimestampType.LOG_APPEND_TIME;
+ long timestamp = msg.getPublishTime();
- ConsumerRecord<K, V> consumerRecord = new ConsumerRecord<>(topic, partition, offset, timestamp,
- timestampType, -1, msg.hasKey() ? msg.getKey().length() : 0, msg.getData().length, key, value);
+ if (msg.getEventTime() > 0) {
+ // If we have Event time, use that in preference
+ timestamp = msg.getEventTime();
+ timestampType = TimestampType.CREATE_TIME;
+ }
- Map<TopicPartition, List<ConsumerRecord<K, V>>> records = new HashMap<>();
- records.put(tp, Lists.newArrayList(consumerRecord));
+ ConsumerRecord<K, V> consumerRecord = new ConsumerRecord<>(topic, partition, offset, timestamp,
+ timestampType, -1, msg.hasKey() ? msg.getKey().length() : 0, msg.getData().length, key, value);
+
+ records.computeIfAbsent(tp, k -> new ArrayList<>()).add(consumerRecord);
+
+ // Update last offset seen by application
+ lastReceivedOffset.put(tp, offset);
+
+ // Check if we have an item already available
+ item = receivedMessages.poll(0, TimeUnit.MILLISECONDS);
+ }
+
+ if (isAutoCommit) {
+ // Commit the offset of previously dequeued messages
+ commitAsync();
+ }
- // Update last offset seen by application
- lastReceivedOffset.put(tp, offset);
return new ConsumerRecords<>(records);
} catch (InterruptedException e) {
throw new RuntimeException(e);
diff --git a/pulsar-client-kafka-compat/pulsar-client-kafka/src/test/java/org/apache/pulsar/client/kafka/compat/tests/KafkaConsumerTest.java b/pulsar-client-kafka-compat/pulsar-client-kafka/src/test/java/org/apache/pulsar/client/kafka/compat/tests/KafkaConsumerTest.java
index 20c3888de..a0218b36b 100644
--- a/pulsar-client-kafka-compat/pulsar-client-kafka/src/test/java/org/apache/pulsar/client/kafka/compat/tests/KafkaConsumerTest.java
+++ b/pulsar-client-kafka-compat/pulsar-client-kafka/src/test/java/org/apache/pulsar/client/kafka/compat/tests/KafkaConsumerTest.java
@@ -27,6 +27,7 @@
import java.util.List;
import java.util.Map;
import java.util.Properties;
+import java.util.concurrent.atomic.AtomicInteger;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecords;
@@ -82,13 +83,14 @@ public void testSimpleConsumer() throws Exception {
pulsarProducer.send(msg);
}
- for (int i = 0; i < 10; i++) {
+ AtomicInteger received = new AtomicInteger();
+ while (received.get() < 10) {
ConsumerRecords<String, String> records = consumer.poll(100);
- assertEquals(records.count(), 1);
- int idx = i;
records.forEach(record -> {
- assertEquals(record.key(), Integer.toString(idx));
- assertEquals(record.value(), "hello-" + idx);
+ assertEquals(record.key(), Integer.toString(received.get()));
+ assertEquals(record.value(), "hello-" + received.get());
+
+ received.incrementAndGet();
});
consumer.commitSync();
@@ -119,13 +121,14 @@ public void testConsumerAutoCommit() throws Exception {
pulsarProducer.send(msg);
}
- for (int i = 0; i < 10; i++) {
+
+ AtomicInteger received = new AtomicInteger();
+ while (received.get() < 10) {
ConsumerRecords<String, String> records = consumer.poll(100);
- assertEquals(records.count(), 1);
- int idx = i;
records.forEach(record -> {
- assertEquals(record.key(), Integer.toString(idx));
- assertEquals(record.value(), "hello-" + idx);
+ assertEquals(record.key(), Integer.toString(received.get()));
+ assertEquals(record.value(), "hello-" + received.get());
+ received.incrementAndGet();
});
}
@@ -162,18 +165,19 @@ public void testConsumerManualOffsetCommit() throws Exception {
pulsarProducer.send(msg);
}
- for (int i = 0; i < 10; i++) {
+ AtomicInteger received = new AtomicInteger();
+ while (received.get() < 10) {
ConsumerRecords<String, String> records = consumer.poll(100);
- assertEquals(records.count(), 1);
- int idx = i;
records.forEach(record -> {
- assertEquals(record.key(), Integer.toString(idx));
- assertEquals(record.value(), "hello-" + idx);
+ assertEquals(record.key(), Integer.toString(received.get()));
+ assertEquals(record.value(), "hello-" + received.get());
Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
offsets.put(new TopicPartition(record.topic(), record.partition()),
new OffsetAndMetadata(record.offset()));
consumer.commitSync(offsets);
+
+ received.incrementAndGet();
});
}
@@ -225,9 +229,10 @@ public void testPartitions() throws Exception {
consumers.forEach(consumer -> {
int expectedMessaged = N / consumers.size();
- for (int i = 0; i < expectedMessaged; i++) {
+
+ for (int i = 0; i < expectedMessaged;) {
ConsumerRecords<String, String> records = consumer.poll(100);
- assertEquals(records.count(), 1);
+ i += records.count();
}
// No more messages for this consumer
@@ -260,13 +265,14 @@ public void testConsumerSeek() throws Exception {
pulsarProducer.send(msg);
}
- for (int i = 0; i < 10; i++) {
+ AtomicInteger received = new AtomicInteger();
+ while (received.get() < 10) {
ConsumerRecords<String, String> records = consumer.poll(100);
- assertEquals(records.count(), 1);
- int idx = i;
records.forEach(record -> {
- assertEquals(record.key(), Integer.toString(idx));
- assertEquals(record.value(), "hello-" + idx);
+ assertEquals(record.key(), Integer.toString(received.get()));
+ assertEquals(record.value(), "hello-" + received.get());
+
+ received.incrementAndGet();
});
consumer.commitSync();
@@ -277,13 +283,14 @@ public void testConsumerSeek() throws Exception {
Thread.sleep(500);
// Messages should be available again
- for (int i = 0; i < 10; i++) {
+ received.set(0);
+ while (received.get() < 10) {
ConsumerRecords<String, String> records = consumer.poll(100);
- assertEquals(records.count(), 1);
- int idx = i;
records.forEach(record -> {
- assertEquals(record.key(), Integer.toString(idx));
- assertEquals(record.value(), "hello-" + idx);
+ assertEquals(record.key(), Integer.toString(received.get()));
+ assertEquals(record.value(), "hello-" + received.get());
+
+ received.incrementAndGet();
});
consumer.commitSync();
@@ -314,13 +321,14 @@ public void testConsumerSeekToEnd() throws Exception {
pulsarProducer.send(msg);
}
- for (int i = 0; i < 10; i++) {
+ AtomicInteger received = new AtomicInteger();
+ while (received.get() < 10) {
ConsumerRecords<String, String> records = consumer.poll(100);
- assertEquals(records.count(), 1);
- int idx = i;
records.forEach(record -> {
- assertEquals(record.key(), Integer.toString(idx));
- assertEquals(record.value(), "hello-" + idx);
+ assertEquals(record.key(), Integer.toString(received.get()));
+ assertEquals(record.value(), "hello-" + received.get());
+
+ received.incrementAndGet();
});
consumer.commitSync();
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services