You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2018/02/28 19:17:26 UTC
[incubator-pulsar] branch master updated: Add Type information to
Pulsar client interface (#987)
This is an automated email from the ASF dual-hosted git repository.
mmerli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 3e76ba2 Add Type information to Pulsar client interface (#987)
3e76ba2 is described below
commit 3e76ba2508ec42e642a70cae09970040b41eab7b
Author: Dave Rusek <da...@gmail.com>
AuthorDate: Wed Feb 28 12:17:24 2018 -0700
Add Type information to Pulsar client interface (#987)
* Add type information to Consumer
* Typed Producer and compiling tests
* Typed consumers
* Typed Client with typed consumers and producers
* Wrap reader and writer configurations in typed variants
* remove type annotation from publisher impl
* Add license headers to new files.
* Remove type from PulsarClient
* Type the consumer on the content not the message
* Fix generic type for Reader config
* Remove type information from PartitionedProducerImpl
* Rename Codec to Schema and add javadoc for client interface
* Make the correct api calls
* Fix incorrectly merged test
* WIP: BROKEN REPLICATOR
* unwrap producer
* Passing tests
* Fix broken merge
* wip
* wip
---
.../pulsar/broker/service/AbstractReplicator.java | 6 +-
.../pulsar/broker/service/BrokerService.java | 1 +
.../nonpersistent/NonPersistentReplicator.java | 2 +-
.../service/persistent/PersistentReplicator.java | 2 +-
.../org/apache/pulsar/client/api/RawReader.java | 5 +-
.../apache/pulsar/client/impl/RawReaderImpl.java | 13 +-
.../broker/auth/MockedPulsarServiceBaseTest.java | 1 +
.../broker/service/PersistentFailoverE2ETest.java | 4 +-
.../pulsar/broker/service/PersistentTopicTest.java | 13 +-
.../pulsar/broker/service/ReplicatorTest.java | 14 +-
.../client/api/DispatcherBlockConsumerTest.java | 61 ++++-----
.../client/api/SimpleProducerConsumerStatTest.java | 4 +-
.../client/api/SimpleProducerConsumerTest.java | 4 +-
.../client/impl/BrokerClientIntegrationTest.java | 2 +-
.../PerMessageUnAcknowledgedRedeliveryTest.java | 8 +-
.../pulsar/client/impl/TopicsConsumerImplTest.java | 10 +-
.../admin/internal/PersistentTopicsImpl.java | 5 +-
.../clients/consumer/PulsarKafkaConsumer.java | 6 +-
.../org/apache/pulsar/client/api/Consumer.java | 16 +--
.../apache/pulsar/client/api/ConsumerBuilder.java | 38 +++---
.../pulsar/client/api/ConsumerConfiguration.java | 6 +-
.../java/org/apache/pulsar/client/api/Message.java | 5 +-
.../apache/pulsar/client/api/MessageBuilder.java | 40 ++++--
.../apache/pulsar/client/api/MessageListener.java | 6 +-
.../org/apache/pulsar/client/api/Producer.java | 16 +--
.../apache/pulsar/client/api/ProducerBuilder.java | 48 +++----
.../org/apache/pulsar/client/api/PulsarClient.java | 71 ++++++++--
.../java/org/apache/pulsar/client/api/Reader.java | 8 +-
.../apache/pulsar/client/api/ReaderBuilder.java | 24 ++--
.../pulsar/client/api/ReaderConfiguration.java | 10 +-
.../apache/pulsar/client/api/ReaderListener.java | 6 +-
.../java/org/apache/pulsar/client/api/Schema.java | 36 ++---
.../apache/pulsar/client/impl/ConsumerBase.java | 35 ++---
.../pulsar/client/impl/ConsumerBuilderImpl.java | 57 ++++----
.../apache/pulsar/client/impl/ConsumerImpl.java | 68 +++++-----
.../pulsar/client/impl/MessageBuilderImpl.java | 37 +++--
.../org/apache/pulsar/client/impl/MessageImpl.java | 37 +++--
.../client/impl/PartitionedConsumerImpl.java | 71 +++++-----
.../client/impl/PartitionedProducerImpl.java | 17 +--
.../client/impl/PatternTopicsConsumerImpl.java | 14 +-
.../apache/pulsar/client/impl/ProducerBase.java | 23 ++--
.../pulsar/client/impl/ProducerBuilderImpl.java | 59 ++++----
.../apache/pulsar/client/impl/ProducerImpl.java | 7 +-
.../pulsar/client/impl/PulsarClientImpl.java | 149 +++++++++++++--------
.../pulsar/client/impl/ReaderBuilderImpl.java | 40 +++---
.../org/apache/pulsar/client/impl/ReaderImpl.java | 34 ++---
.../pulsar/client/impl/TopicMessageImpl.java | 5 +
.../pulsar/client/impl/TopicsConsumerImpl.java | 57 ++++----
.../impl/conf/ConsumerConfigurationData.java | 16 +--
.../client/impl/conf/ReaderConfigurationData.java | 8 +-
.../java/org/apache/pulsar/storm/PulsarBolt.java | 4 +-
.../apache/pulsar/storm/TupleToMessageMapper.java | 2 +-
.../pulsar/testclient/LoadSimulationClient.java | 2 +-
.../pulsar/testclient/PerformanceConsumer.java | 4 +-
.../pulsar/testclient/PerformanceProducer.java | 8 +-
.../pulsar/testclient/PerformanceReader.java | 2 +-
.../apache/pulsar/websocket/ConsumerHandler.java | 11 +-
.../apache/pulsar/websocket/ProducerHandler.java | 4 +-
.../org/apache/pulsar/websocket/ReaderHandler.java | 10 +-
59 files changed, 706 insertions(+), 566 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java
index 1919008..d3d15d9 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java
@@ -47,7 +47,7 @@ public abstract class AbstractReplicator {
protected volatile ProducerImpl producer;
protected final int producerQueueSize;
- protected final ProducerBuilder producerBuilder;
+ protected final ProducerBuilder<byte[]> producerBuilder;
protected final Backoff backOff = new Backoff(100, TimeUnit.MILLISECONDS, 1, TimeUnit.MINUTES, 0 ,TimeUnit.MILLISECONDS);
@@ -80,7 +80,7 @@ public abstract class AbstractReplicator {
STATE_UPDATER.set(this, State.Stopped);
}
- protected abstract void readEntries(org.apache.pulsar.client.api.Producer producer);
+ protected abstract void readEntries(org.apache.pulsar.client.api.Producer<byte[]> producer);
protected abstract Position getReplicatorReadPosition();
@@ -231,7 +231,7 @@ public abstract class AbstractReplicator {
* Therefore, replicator can't be started on root-partition topic which can internally create multiple partitioned
* producers.
*
- * @param topicName
+ * @param topic
* @param brokerService
*/
private void validatePartitionedTopic(String topic, BrokerService brokerService) throws NamingException {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
index 575f1ec..1910680 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
@@ -79,6 +79,7 @@ import org.apache.pulsar.broker.zookeeper.aspectj.ClientCnxnAspect;
import org.apache.pulsar.broker.zookeeper.aspectj.ClientCnxnAspect.EventListner;
import org.apache.pulsar.broker.zookeeper.aspectj.ClientCnxnAspect.EventType;
import org.apache.pulsar.client.api.ClientConfiguration;
+import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.impl.PulsarClientImpl;
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentReplicator.java
index 74af8b6..42b99ad 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentReplicator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentReplicator.java
@@ -59,7 +59,7 @@ public class NonPersistentReplicator extends AbstractReplicator implements Repli
}
@Override
- protected void readEntries(Producer producer) {
+ protected void readEntries(Producer<byte[]> producer) {
this.producer = (ProducerImpl) producer;
if (STATE_UPDATER.compareAndSet(this, State.Starting, State.Started)) {
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
index 52ed4b4..99804df 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java
@@ -105,7 +105,7 @@ public class PersistentReplicator extends AbstractReplicator implements Replicat
}
@Override
- protected void readEntries(org.apache.pulsar.client.api.Producer producer) {
+ protected void readEntries(org.apache.pulsar.client.api.Producer<byte[]> producer) {
// Rewind the cursor to be sure to read again all non-acked messages sent while restarting
cursor.rewind();
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawReader.java b/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawReader.java
index 15b03be..7f7cd40 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawReader.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/client/api/RawReader.java
@@ -20,8 +20,6 @@ package org.apache.pulsar.client.api;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
-import org.apache.pulsar.client.api.MessageId;
-import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.client.impl.RawReaderImpl;
@@ -32,8 +30,9 @@ public interface RawReader {
/**
* Create a raw reader for a topic.
*/
+
public static CompletableFuture<RawReader> create(PulsarClient client, String topic, String subscription) {
- CompletableFuture<Consumer> future = new CompletableFuture<>();
+ CompletableFuture<Consumer<byte[]>> future = new CompletableFuture<>();
RawReader r = new RawReaderImpl((PulsarClientImpl)client, topic, subscription, future);
return future.thenCompose((consumer) -> r.seekAsync(MessageId.earliest)).thenApply((ignore) -> r);
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java
index a048a5b..75cbbbc 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/client/impl/RawReaderImpl.java
@@ -30,6 +30,7 @@ import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.RawMessage;
import org.apache.pulsar.client.api.RawReader;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
@@ -43,12 +44,12 @@ import io.netty.buffer.ByteBuf;
public class RawReaderImpl implements RawReader {
final static int DEFAULT_RECEIVER_QUEUE_SIZE = 1000;
- private final ConsumerConfigurationData consumerConfiguration;
+ private final ConsumerConfigurationData<byte[]> consumerConfiguration;
private RawConsumerImpl consumer;
public RawReaderImpl(PulsarClientImpl client, String topic, String subscription,
- CompletableFuture<Consumer> consumerFuture) {
- consumerConfiguration = new ConsumerConfigurationData();
+ CompletableFuture<Consumer<byte[]>> consumerFuture) {
+ consumerConfiguration = new ConsumerConfigurationData<>();
consumerConfiguration.getTopicNames().add(topic);
consumerConfiguration.setSubscriptionName(subscription);
consumerConfiguration.setSubscriptionType(SubscriptionType.Exclusive);
@@ -83,14 +84,14 @@ public class RawReaderImpl implements RawReader {
return consumer.getLastMessageIdAsync();
}
- static class RawConsumerImpl extends ConsumerImpl {
+ static class RawConsumerImpl extends ConsumerImpl<byte[]> {
final BlockingQueue<RawMessageAndCnx> incomingRawMessages;
final Queue<CompletableFuture<RawMessage>> pendingRawReceives;
RawConsumerImpl(PulsarClientImpl client, ConsumerConfigurationData conf,
- CompletableFuture<Consumer> consumerFuture) {
+ CompletableFuture<Consumer<byte[]>> consumerFuture) {
super(client, conf.getSingleTopic(), conf, client.externalExecutorProvider().getExecutor(), -1,
- consumerFuture, SubscriptionMode.Durable, MessageId.earliest);
+ consumerFuture, SubscriptionMode.Durable, MessageId.earliest, Schema.IDENTITY);
incomingRawMessages = new GrowableArrayBlockingQueue<>();
pendingRawReceives = new ConcurrentLinkedQueue<>();
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
index d189559..c0e3839 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java
@@ -43,6 +43,7 @@ import org.apache.pulsar.broker.ServiceConfiguration;
import org.apache.pulsar.broker.namespace.NamespaceService;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.Authentication;
+import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.zookeeper.ZooKeeperClientFactory;
import org.apache.pulsar.zookeeper.ZookeeperClientFactoryImpl;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentFailoverE2ETest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentFailoverE2ETest.java
index fa4da0b..dda0bc0 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentFailoverE2ETest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentFailoverE2ETest.java
@@ -584,8 +584,8 @@ public class PersistentFailoverE2ETest extends BrokerTestBase {
producer.close();
// two consumers subscribe at almost the same time
- CompletableFuture<Consumer> subscribeFuture2 = pulsarClient.subscribeAsync(topicName, subName, consumerConf2);
- CompletableFuture<Consumer> subscribeFuture1 = pulsarClient.subscribeAsync(topicName, subName, consumerConf1);
+ CompletableFuture<Consumer<byte[]>> subscribeFuture2 = pulsarClient.subscribeAsync(topicName, subName, consumerConf2);
+ CompletableFuture<Consumer<byte[]>> subscribeFuture1 = pulsarClient.subscribeAsync(topicName, subName, consumerConf1);
// wait for all messages to be dequeued
int retry = 20;
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
index c1b44c0..655ccab 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/PersistentTopicTest.java
@@ -86,6 +86,7 @@ import org.apache.pulsar.broker.service.persistent.PersistentReplicator;
import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
import org.apache.pulsar.broker.service.persistent.PersistentTopic;
import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
@@ -1213,14 +1214,22 @@ public class PersistentTopicTest {
.createProducerAsync(any(ProducerConfigurationData.class));
replicator.startProducer();
- verify(clientImpl).createProducerAsync(any(ProducerConfigurationData.class));
+ verify(clientImpl)
+ .createProducerAsync(
+ any(ProducerConfigurationData.class),
+ any(Schema.class)
+ );
replicator.disconnect(false);
replicator.disconnect(false);
replicator.startProducer();
- verify(clientImpl, Mockito.times(2)).createProducerAsync(any(ProducerConfigurationData.class));
+ verify(clientImpl, Mockito.times(2))
+ .createProducerAsync(
+ any(ProducerConfigurationData.class),
+ any(Schema.class)
+ );
}
@Test
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
index 6f9a51c..170ee2d 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatorTest.java
@@ -24,6 +24,8 @@ import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
+import com.scurrilous.circe.checksum.Crc32cIntChecksum;
+import io.netty.buffer.ByteBuf;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
import java.util.List;
@@ -37,7 +39,6 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
-
import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCursorCallback;
import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntriesCallback;
import org.apache.bookkeeper.mledger.Entry;
@@ -58,6 +59,7 @@ import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.RawMessage;
import org.apache.pulsar.client.api.RawReader;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.ProducerImpl;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
@@ -80,10 +82,6 @@ import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
import org.testng.collections.Lists;
-import com.scurrilous.circe.checksum.Crc32cIntChecksum;
-
-import io.netty.buffer.ByteBuf;
-
/**
* Starts 2 brokers that are in 2 different clusters
*/
@@ -244,7 +242,11 @@ public class ReplicatorTest extends ReplicatorTestBase {
}
Thread.sleep(3000);
- Mockito.verify(pulsarClient, Mockito.times(1)).createProducerAsync(Mockito.any(ProducerConfigurationData.class));
+ Mockito.verify(pulsarClient, Mockito.times(1))
+ .createProducerAsync(
+ Mockito.any(ProducerConfigurationData.class),
+ Mockito.any(Schema.class)
+ );
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DispatcherBlockConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DispatcherBlockConsumerTest.java
index 3167006..f2c63e3 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DispatcherBlockConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/DispatcherBlockConsumerTest.java
@@ -27,6 +27,7 @@ import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
import java.lang.reflect.Field;
+import java.util.*;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
@@ -109,10 +110,10 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
ConsumerConfiguration conf = new ConsumerConfiguration();
conf.setReceiverQueueSize(receiverQueueSize);
conf.setSubscriptionType(SubscriptionType.Shared);
- Consumer consumer1 = pulsarClient.subscribe(topicName, subscriberName, conf);
- Consumer consumer2 = pulsarClient.subscribe(topicName, subscriberName, conf);
- Consumer consumer3 = pulsarClient.subscribe(topicName, subscriberName, conf);
- Consumer[] consumers = { consumer1, consumer2, consumer3 };
+ Consumer<byte[]> consumer1 = pulsarClient.subscribe(topicName, subscriberName, conf);
+ Consumer<byte[]> consumer2 = pulsarClient.subscribe(topicName, subscriberName, conf);
+ Consumer<byte[]> consumer3 = pulsarClient.subscribe(topicName, subscriberName, conf);
+ List<Consumer<byte[]>> consumers = Lists.newArrayList(consumer1, consumer2, consumer3);
ProducerConfiguration producerConf = new ProducerConfiguration();
@@ -130,9 +131,9 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
Map<Message, Consumer> messages = Maps.newHashMap();
for (int i = 0; i < 3; i++) {
for (int j = 0; j < totalProducedMsgs; j++) {
- msg = consumers[i].receive(500, TimeUnit.MILLISECONDS);
+ msg = consumers.get(i).receive(500, TimeUnit.MILLISECONDS);
if (msg != null) {
- messages.put(msg, consumers[i]);
+ messages.put(msg, consumers.get(i));
} else {
break;
}
@@ -158,13 +159,13 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
// expecting messages which are not received
int expectedRemainingMessages = totalProducedMsgs - messages.size();
CountDownLatch latch = new CountDownLatch(expectedRemainingMessages);
- for (int i = 0; i < consumers.length; i++) {
+ for (int i = 0; i < consumers.size(); i++) {
final int consumerCount = i;
for (int j = 0; j < totalProducedMsgs; j++) {
- consumers[i].receiveAsync().thenAccept(m -> {
+ consumers.get(i).receiveAsync().thenAccept(m -> {
result.add(m.getMessageId());
try {
- consumers[consumerCount].acknowledge(m);
+ consumers.get(consumerCount).acknowledge(m);
} catch (PulsarClientException e) {
fail("failed to ack msg", e);
}
@@ -179,7 +180,7 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
assertTrue(result.size() >= expectedRemainingMessages);
producer.close();
- Arrays.asList(consumers).forEach(c -> {
+ consumers.forEach(c -> {
try {
c.close();
} catch (PulsarClientException e) {
@@ -219,7 +220,7 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
ConsumerImpl consumer1 = (ConsumerImpl) pulsarClient.subscribe(topicName, subscriberName, conf);
ConsumerImpl consumer2 = (ConsumerImpl) pulsarClient.subscribe(topicName, subscriberName, conf);
ConsumerImpl consumer3 = (ConsumerImpl) pulsarClient.subscribe(topicName, subscriberName, conf);
- ConsumerImpl[] consumers = { consumer1, consumer2, consumer3 };
+ List<ConsumerImpl<byte[]>> consumers = Lists.newArrayList(consumer1, consumer2, consumer3);
ProducerConfiguration producerConf = new ProducerConfiguration();
@@ -237,9 +238,9 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
Multimap<ConsumerImpl, MessageId> messages = ArrayListMultimap.create();
for (int i = 0; i < 3; i++) {
for (int j = 0; j < totalProducedMsgs; j++) {
- msg = consumers[i].receive(500, TimeUnit.MILLISECONDS);
+ msg = consumers.get(i).receive(500, TimeUnit.MILLISECONDS);
if (msg != null) {
- messages.put(consumers[i], msg.getMessageId());
+ messages.put(consumers.get(i), msg.getMessageId());
log.info("Received message: " + new String(msg.getData()));
} else {
break;
@@ -260,13 +261,13 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
// all messages
Queue<MessageId> result = Queues.newConcurrentLinkedQueue();
CountDownLatch latch = new CountDownLatch(totalProducedMsgs);
- for (int i = 0; i < consumers.length; i++) {
+ for (int i = 0; i < consumers.size(); i++) {
final int consumerCount = i;
for (int j = 0; j < totalProducedMsgs; j++) {
- consumers[i].receiveAsync().thenAccept(m -> {
+ consumers.get(i).receiveAsync().thenAccept(m -> {
result.add(m.getMessageId());
try {
- consumers[consumerCount].acknowledge(m);
+ consumers.get(consumerCount).acknowledge(m);
} catch (PulsarClientException e) {
fail("failed to ack msg", e);
}
@@ -280,7 +281,7 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
// total received-messages should match to produced messages (it may have duplicate messages)
assertTrue(result.size() >= totalProducedMsgs);
producer.close();
- Arrays.asList(consumers).forEach(c -> {
+ consumers.forEach(c -> {
try {
c.close();
} catch (PulsarClientException e) {
@@ -410,7 +411,7 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
ConsumerImpl consumer1 = (ConsumerImpl) pulsarClient.subscribe(topicName, subscriberName, conf);
ConsumerImpl consumer2 = (ConsumerImpl) pulsarClient.subscribe(topicName, subscriberName, conf);
ConsumerImpl consumer3 = (ConsumerImpl) pulsarClient.subscribe(topicName, subscriberName, conf);
- ConsumerImpl[] consumers = { consumer1, consumer2, consumer3 };
+ List<ConsumerImpl<byte[]>> consumers = Lists.newArrayList( consumer1, consumer2, consumer3 );
ProducerConfiguration producerConf = new ProducerConfiguration();
@@ -428,7 +429,7 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
Set<MessageId> messages = Sets.newHashSet();
for (int i = 0; i < 3; i++) {
for (int j = 0; j < totalProducedMsgs; j++) {
- msg = consumers[i].receive(500, TimeUnit.MILLISECONDS);
+ msg = consumers.get(i).receive(500, TimeUnit.MILLISECONDS);
if (msg != null) {
messages.add(msg.getMessageId());
log.info("Received message: " + new String(msg.getData()));
@@ -443,7 +444,7 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
assertEquals(totalConsumedMsgs, unackMsgAllowed, 3 * receiverQueueSize);
// trigger redelivery
- Arrays.asList(consumers).forEach(c -> {
+ consumers.forEach(c -> {
c.redeliverUnacknowledgedMessages();
});
@@ -454,10 +455,10 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
Map<ConsumerImpl, Set<MessageId>> messages1 = Maps.newHashMap();
for (int i = 0; i < 3; i++) {
for (int j = 0; j < totalProducedMsgs; j++) {
- msg = consumers[i].receive(500, TimeUnit.MILLISECONDS);
+ msg = consumers.get(i).receive(500, TimeUnit.MILLISECONDS);
if (msg != null) {
- messages1.putIfAbsent(consumers[i], Sets.newHashSet());
- messages1.get(consumers[i]).add(msg.getMessageId());
+ messages1.putIfAbsent(consumers.get(i), Sets.newHashSet());
+ messages1.get(consumers.get(i)).add(msg.getMessageId());
log.info("Received message: " + new String(msg.getData()));
} else {
break;
@@ -466,7 +467,7 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
}
Set<MessageId> result = Sets.newHashSet();
- messages1.values().forEach(s -> result.addAll(s));
+ messages1.values().forEach(result::addAll);
// check all unacked messages have been redelivered
@@ -483,20 +484,20 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
});
});
- messages1.values().forEach(s -> result.addAll(s));
+ messages1.values().forEach(result::addAll);
// try to consume remaining messages
int remainingMessages = totalProducedMsgs - result.size();
// try to consume remaining messages: broker may take time to deliver so, retry multiple time to consume
// all messages
CountDownLatch latch = new CountDownLatch(remainingMessages);
Queue<MessageId> consumedMessages = Queues.newConcurrentLinkedQueue();
- for (int i = 0; i < consumers.length; i++) {
+ for (int i = 0; i < consumers.size(); i++) {
final int counsumerIndex = i;
for (int j = 0; j < remainingMessages; j++) {
- consumers[i].receiveAsync().thenAccept(m -> {
+ consumers.get(i).receiveAsync().thenAccept(m -> {
consumedMessages.add(m.getMessageId());
try {
- consumers[counsumerIndex].acknowledge(m);
+ consumers.get(counsumerIndex).acknowledge(m);
} catch (PulsarClientException e) {
fail("failed to ack", e);
}
@@ -509,7 +510,7 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
// total received-messages should match remaining messages excluding duplicate
assertTrue(consumedMessages.size() >= remainingMessages);
producer.close();
- Arrays.asList(consumers).forEach(c -> {
+ consumers.forEach(c -> {
try {
c.close();
} catch (PulsarClientException e) {
@@ -782,7 +783,7 @@ public class DispatcherBlockConsumerTest extends ProducerConsumerBase {
* (2) However, other subscription2 should still be able to consume messages until it reaches to
* maxUnAckPerDispatcher limit
**/
- ConsumerImpl consumerSub2 = (ConsumerImpl) pulsarClient.subscribe(topicName, subscriberName2, conf);
+ ConsumerImpl<byte[]> consumerSub2 = (ConsumerImpl) pulsarClient.subscribe(topicName, subscriberName2, conf);
Set<MessageId> messages2 = Sets.newHashSet();
for (int j = 0; j < totalProducedMsgs; j++) {
msg = consumerSub2.receive(100, TimeUnit.MILLISECONDS);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerStatTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerStatTest.java
index a13ac3f..0bc78da 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerStatTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerStatTest.java
@@ -346,7 +346,7 @@ public class SimpleProducerConsumerStatTest extends ProducerConsumerBase {
int batchSize = 5;
ConsumerConfiguration consumerConf = new ConsumerConfiguration();
consumerConf.setSubscriptionType(SubscriptionType.Exclusive);
- Consumer consumer = pulsarClient.subscribe(topicName, "my-subscriber-name", consumerConf);
+ Consumer<byte[]> consumer = pulsarClient.subscribe(topicName, "my-subscriber-name", consumerConf);
ProducerConfiguration producerConf = new ProducerConfiguration();
producerConf.setBatchingMaxMessages(batchSize);
producerConf.setBatchingEnabled(true);
@@ -359,7 +359,7 @@ public class SimpleProducerConsumerStatTest extends ProducerConsumerBase {
while (runTest.get()) {
r.acquire();
producer.sendAsync("Hello World".getBytes());
- consumer.receiveAsync().thenAccept(message -> consumer.acknowledgeAsync(message));
+ consumer.receiveAsync().thenAccept(consumer::acknowledgeAsync);
}
});
t1.start();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
index 33d1063..3aff2a7 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerTest.java
@@ -399,7 +399,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
}
try {
- Producer producer = pulsarClient.createProducer("persistent://my-property/use/my-ns/my-topic7", null);
+ Producer producer = pulsarClient.createProducer("persistent://my-property/use/my-ns/my-topic7", (ProducerConfiguration) null);
Assert.fail("should fail");
} catch (PulsarClientException e) {
Assert.assertTrue(e instanceof PulsarClientException.InvalidConfigurationException);
@@ -437,7 +437,7 @@ public class SimpleProducerConsumerTest extends ProducerConsumerBase {
try {
Consumer consumer = pulsarClient.subscribe("persistent://my-property/use/my-ns/my-topic7",
- "my-subscriber-name", null);
+ "my-subscriber-name", (ConsumerConfiguration) null);
Assert.fail("Should fail");
} catch (PulsarClientException e) {
Assert.assertTrue(e instanceof PulsarClientException.InvalidConfigurationException);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java
index 8cf5618..fe99762 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/BrokerClientIntegrationTest.java
@@ -602,7 +602,7 @@ public class BrokerClientIntegrationTest extends ProducerConsumerBase {
ClientCnx cnx = producer.cnx();
assertTrue(cnx.channel().isActive());
ExecutorService executor = Executors.newFixedThreadPool(concurrentLookupRequests);
- List<CompletableFuture<Producer>> futures = Lists.newArrayList();
+ List<CompletableFuture<Producer<byte[]>>> futures = Lists.newArrayList();
final int totalProducers = 10;
CountDownLatch latch = new CountDownLatch(totalProducers);
final ProducerConfiguration config1 = new ProducerConfiguration();
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PerMessageUnAcknowledgedRedeliveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PerMessageUnAcknowledgedRedeliveryTest.java
index e9cecbb..1d89b56 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PerMessageUnAcknowledgedRedeliveryTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PerMessageUnAcknowledgedRedeliveryTest.java
@@ -347,8 +347,8 @@ public class PerMessageUnAcknowledgedRedeliveryTest extends BrokerTestBase {
assertEquals(size, 0);
}
- private static long getUnackedMessagesCountInPartitionedConsumer(Consumer c) {
- PartitionedConsumerImpl pc = (PartitionedConsumerImpl) c;
+ private static long getUnackedMessagesCountInPartitionedConsumer(Consumer<byte[]> c) {
+ PartitionedConsumerImpl<byte[]> pc = (PartitionedConsumerImpl) c;
return pc.getUnAckedMessageTracker().size() + pc.getConsumers().stream()
.mapToLong(consumer -> consumer.getUnAckedMessageTracker().size()).sum();
}
@@ -374,7 +374,7 @@ public class PerMessageUnAcknowledgedRedeliveryTest extends BrokerTestBase {
conf.setReceiverQueueSize(50);
conf.setAckTimeout(ackTimeOutMillis, TimeUnit.MILLISECONDS);
conf.setSubscriptionType(SubscriptionType.Shared);
- Consumer consumer = pulsarClient.subscribe(topicName, subscriptionName, conf);
+ Consumer<byte[]> consumer = pulsarClient.subscribe(topicName, subscriptionName, conf);
// 3. producer publish messages
for (int i = 0; i < totalMessages / 3; i++) {
@@ -419,7 +419,7 @@ public class PerMessageUnAcknowledgedRedeliveryTest extends BrokerTestBase {
// 7. Simulate ackTimeout
((PartitionedConsumerImpl) consumer).getUnAckedMessageTracker().toggle();
- ((PartitionedConsumerImpl) consumer).getConsumers().forEach(c -> c.getUnAckedMessageTracker().toggle());
+ ((PartitionedConsumerImpl<byte[]>) consumer).getConsumers().forEach(c -> c.getUnAckedMessageTracker().toggle());
// 8. producer publish more messages
for (int i = 0; i < totalMessages / 3; i++) {
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java
index 2aa386e..cf741a4 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java
@@ -211,17 +211,17 @@ public class TopicsConsumerImplTest extends ProducerConsumerBase {
admin.persistentTopics().createPartitionedTopic(topicName3, 3);
// 1. producer connect
- Producer producer1 = pulsarClient.newProducer().topic(topicName1)
+ Producer<byte[]> producer1 = pulsarClient.newProducer().topic(topicName1)
.create();
- Producer producer2 = pulsarClient.newProducer().topic(topicName2)
+ Producer<byte[]> producer2 = pulsarClient.newProducer().topic(topicName2)
.messageRoutingMode(org.apache.pulsar.client.api.MessageRoutingMode.RoundRobinPartition)
.create();
- Producer producer3 = pulsarClient.newProducer().topic(topicName3)
+ Producer<byte[]> producer3 = pulsarClient.newProducer().topic(topicName3)
.messageRoutingMode(org.apache.pulsar.client.api.MessageRoutingMode.RoundRobinPartition)
.create();
// 2. Create consumer
- Consumer consumer = pulsarClient.newConsumer()
+ Consumer<byte[]> consumer = pulsarClient.newConsumer()
.topics(topicNames)
.subscriptionName(subscriptionName)
.subscriptionType(SubscriptionType.Shared)
@@ -368,7 +368,7 @@ public class TopicsConsumerImplTest extends ProducerConsumerBase {
// 8. Simulate ackTimeout
((TopicsConsumerImpl) consumer).getUnAckedMessageTracker().toggle();
- ((TopicsConsumerImpl) consumer).getConsumers().forEach(c -> c.getUnAckedMessageTracker().toggle());
+ ((TopicsConsumerImpl<byte[]>) consumer).getConsumers().forEach(c -> c.getUnAckedMessageTracker().toggle());
// 9. producer publish more messages
for (int i = 0; i < totalMessages / 3; i++) {
diff --git a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PersistentTopicsImpl.java b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PersistentTopicsImpl.java
index 92bd242..cf5b47e 100644
--- a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PersistentTopicsImpl.java
+++ b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PersistentTopicsImpl.java
@@ -47,6 +47,7 @@ import org.apache.pulsar.client.admin.PulsarAdminException.NotFoundException;
import org.apache.pulsar.client.api.Authentication;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.MessageIdImpl;
import org.apache.pulsar.client.impl.MessageImpl;
import org.apache.pulsar.common.api.Commands;
@@ -761,7 +762,7 @@ public class PersistentTopicsImpl extends BaseResource implements PersistentTopi
}
}
- return Lists.newArrayList(new MessageImpl(msgId, properties, data));
+ return Lists.newArrayList(new MessageImpl<>(msgId, properties, data, Schema.IDENTITY));
} finally {
if (stream != null) {
stream.close();
@@ -786,7 +787,7 @@ public class PersistentTopicsImpl extends BaseResource implements PersistentTopi
properties.put(entry.getKey(), entry.getValue());
}
}
- ret.add(new MessageImpl(batchMsgId, properties, singleMessagePayload));
+ ret.add(new MessageImpl<>(batchMsgId, properties, singleMessagePayload, Schema.IDENTITY));
} catch (Exception ex) {
log.error("Exception occured while trying to get BatchMsgId: {}", batchMsgId, ex);
}
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 0a4b9d2..31c9921 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
@@ -195,7 +195,7 @@ public class PulsarKafkaConsumer<K, V> implements Consumer<K, V>, MessageListene
@Override
public void subscribe(Collection<String> topics, ConsumerRebalanceListener callback) {
- List<CompletableFuture<org.apache.pulsar.client.api.Consumer>> futures = new ArrayList<>();
+ List<CompletableFuture<org.apache.pulsar.client.api.Consumer<byte[]>>> futures = new ArrayList<>();
List<TopicPartition> topicPartitions = new ArrayList<>();
try {
@@ -212,7 +212,7 @@ public class PulsarKafkaConsumer<K, V> implements Consumer<K, V>, MessageListene
conf.setConsumerName(ConsumerName.generateRandomName());
for (int i = 0; i < numberOfPartitions; i++) {
String partitionName = TopicName.get(topic).getPartition(i).toString();
- CompletableFuture<org.apache.pulsar.client.api.Consumer> future = client
+ CompletableFuture<org.apache.pulsar.client.api.Consumer<byte[]>> future = client
.subscribeAsync(partitionName, groupId, conf);
int partitionIndex = i;
TopicPartition tp = new TopicPartition(topic, partitionIndex);
@@ -222,7 +222,7 @@ public class PulsarKafkaConsumer<K, V> implements Consumer<K, V>, MessageListene
}
} else {
// Topic has a single partition
- CompletableFuture<org.apache.pulsar.client.api.Consumer> future = client.subscribeAsync(topic,
+ CompletableFuture<org.apache.pulsar.client.api.Consumer<byte[]>> future = client.subscribeAsync(topic,
groupId, conf);
TopicPartition tp = new TopicPartition(topic, 0);
future.thenAccept(consumer -> consumers.putIfAbsent(tp, consumer));
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/api/Consumer.java b/pulsar-client/src/main/java/org/apache/pulsar/client/api/Consumer.java
index 2da67eb..a1dd628 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/api/Consumer.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/api/Consumer.java
@@ -29,7 +29,7 @@ import org.apache.pulsar.client.impl.ConsumerStats;
*
*
*/
-public interface Consumer extends Closeable {
+public interface Consumer<T> extends Closeable {
/**
* Get a topic for the consumer
@@ -72,7 +72,7 @@ public interface Consumer extends Closeable {
* @throws PulsarClientException.InvalidConfigurationException
* if a message listener was defined in the configuration
*/
- Message receive() throws PulsarClientException;
+ Message<T> receive() throws PulsarClientException;
/**
* Receive a single message
@@ -86,7 +86,7 @@ public interface Consumer extends Closeable {
*
* @return {@link CompletableFuture}<{@link Message}> will be completed when message is available
*/
- CompletableFuture<Message> receiveAsync();
+ CompletableFuture<Message<T>> receiveAsync();
/**
* Receive a single message
@@ -102,7 +102,7 @@ public interface Consumer extends Closeable {
* @throws PulsarClientException.InvalidConfigurationException
* if a message listener was defined in the configuration
*/
- Message receive(int timeout, TimeUnit unit) throws PulsarClientException;
+ Message<T> receive(int timeout, TimeUnit unit) throws PulsarClientException;
/**
* Acknowledge the consumption of a single message
@@ -112,7 +112,7 @@ public interface Consumer extends Closeable {
* @throws PulsarClientException.AlreadyClosedException
* if the consumer was already closed
*/
- void acknowledge(Message message) throws PulsarClientException;
+ void acknowledge(Message<?> message) throws PulsarClientException;
/**
* Acknowledge the consumption of a single message, identified by its MessageId
@@ -139,7 +139,7 @@ public interface Consumer extends Closeable {
* @throws PulsarClientException.AlreadyClosedException
* if the consumer was already closed
*/
- void acknowledgeCumulative(Message message) throws PulsarClientException;
+ void acknowledgeCumulative(Message<?> message) throws PulsarClientException;
/**
* Acknowledge the reception of all the messages in the stream up to (and including) the provided message.
@@ -165,7 +165,7 @@ public interface Consumer extends Closeable {
* The {@code Message} to be acknowledged
* @return a future that can be used to track the completion of the operation
*/
- CompletableFuture<Void> acknowledgeAsync(Message message);
+ CompletableFuture<Void> acknowledgeAsync(Message<?> message);
/**
* Asynchronously acknowledge the consumption of a single message
@@ -186,7 +186,7 @@ public interface Consumer extends Closeable {
* The {@code Message} to be cumulatively acknowledged
* @return a future that can be used to track the completion of the operation
*/
- CompletableFuture<Void> acknowledgeCumulativeAsync(Message message);
+ CompletableFuture<Void> acknowledgeCumulativeAsync(Message<?> message);
/**
* Asynchronously Acknowledge the reception of all the messages in the stream up to (and including) the provided
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java
index c487425..d9dcb1c 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java
@@ -32,7 +32,7 @@ import java.util.regex.Pattern;
*
* @since 2.0.0
*/
-public interface ConsumerBuilder extends Serializable, Cloneable {
+public interface ConsumerBuilder<T> extends Serializable, Cloneable {
/**
* Create a copy of the current consumer builder.
@@ -50,7 +50,7 @@ public interface ConsumerBuilder extends Serializable, Cloneable {
* Consumer consumer2 = builder.clone().topic(TOPIC_2).subscribe();
* </pre>
*/
- ConsumerBuilder clone();
+ ConsumerBuilder<T> clone();
/**
* Finalize the {@link Consumer} creation by subscribing to the topic.
@@ -63,7 +63,7 @@ public interface ConsumerBuilder extends Serializable, Cloneable {
* @throws PulsarClientException
* if the the subscribe operation fails
*/
- Consumer subscribe() throws PulsarClientException;
+ Consumer<T> subscribe() throws PulsarClientException;
/**
* Finalize the {@link Consumer} creation by subscribing to the topic in asynchronous mode.
@@ -76,7 +76,7 @@ public interface ConsumerBuilder extends Serializable, Cloneable {
* @throws PulsarClientException
* if the the subscribe operation fails
*/
- CompletableFuture<Consumer> subscribeAsync();
+ CompletableFuture<Consumer<T>> subscribeAsync();
/**
* Specify the topics this consumer will subscribe on.
@@ -84,7 +84,7 @@ public interface ConsumerBuilder extends Serializable, Cloneable {
*
* @param topicNames
*/
- ConsumerBuilder topic(String... topicNames);
+ ConsumerBuilder<T> topic(String... topicNames);
/**
* Specify a list of topics that this consumer will subscribe on.
@@ -92,7 +92,7 @@ public interface ConsumerBuilder extends Serializable, Cloneable {
*
* @param topicNames
*/
- ConsumerBuilder topics(List<String> topicNames);
+ ConsumerBuilder<T> topics(List<String> topicNames);
/**
* Specify a pattern for topics that this consumer will subscribe on.
@@ -120,7 +120,7 @@ public interface ConsumerBuilder extends Serializable, Cloneable {
*
* @param subscriptionName
*/
- ConsumerBuilder subscriptionName(String subscriptionName);
+ ConsumerBuilder<T> subscriptionName(String subscriptionName);
/**
* Set the timeout for unacked messages, truncated to the nearest millisecond. The timeout needs to be greater than
@@ -132,7 +132,7 @@ public interface ConsumerBuilder extends Serializable, Cloneable {
* unit in which the timeout is provided.
* @return {@link ConsumerConfiguration}
*/
- ConsumerBuilder ackTimeout(long ackTimeout, TimeUnit timeUnit);
+ ConsumerBuilder<T> ackTimeout(long ackTimeout, TimeUnit timeUnit);
/**
* Select the subscription type to be used when subscribing to the topic.
@@ -142,7 +142,7 @@ public interface ConsumerBuilder extends Serializable, Cloneable {
* @param subscriptionType
* the subscription type value
*/
- ConsumerBuilder subscriptionType(SubscriptionType subscriptionType);
+ ConsumerBuilder<T> subscriptionType(SubscriptionType subscriptionType);
/**
* Sets a {@link MessageListener} for the consumer
@@ -153,7 +153,7 @@ public interface ConsumerBuilder extends Serializable, Cloneable {
* @param messageListener
* the listener object
*/
- ConsumerBuilder messageListener(MessageListener messageListener);
+ ConsumerBuilder<T> messageListener(MessageListener<T> messageListener);
/**
* Sets a {@link CryptoKeyReader}
@@ -161,7 +161,7 @@ public interface ConsumerBuilder extends Serializable, Cloneable {
* @param cryptoKeyReader
* CryptoKeyReader object
*/
- ConsumerBuilder cryptoKeyReader(CryptoKeyReader cryptoKeyReader);
+ ConsumerBuilder<T> cryptoKeyReader(CryptoKeyReader cryptoKeyReader);
/**
* Sets the ConsumerCryptoFailureAction to the value specified
@@ -169,7 +169,7 @@ public interface ConsumerBuilder extends Serializable, Cloneable {
* @param action
* The consumer action
*/
- ConsumerBuilder cryptoFailureAction(ConsumerCryptoFailureAction action);
+ ConsumerBuilder<T> cryptoFailureAction(ConsumerCryptoFailureAction action);
/**
* Sets the size of the consumer receive queue.
@@ -197,7 +197,7 @@ public interface ConsumerBuilder extends Serializable, Cloneable {
* @param receiverQueueSize
* the new receiver queue size value
*/
- ConsumerBuilder receiverQueueSize(int receiverQueueSize);
+ ConsumerBuilder<T> receiverQueueSize(int receiverQueueSize);
/**
* Set the max total receiver queue size across partitons.
@@ -207,14 +207,14 @@ public interface ConsumerBuilder extends Serializable, Cloneable {
*
* @param maxTotalReceiverQueueSizeAcrossPartitions
*/
- ConsumerBuilder maxTotalReceiverQueueSizeAcrossPartitions(int maxTotalReceiverQueueSizeAcrossPartitions);
+ ConsumerBuilder<T> maxTotalReceiverQueueSizeAcrossPartitions(int maxTotalReceiverQueueSizeAcrossPartitions);
/**
* Set the consumer name.
*
* @param consumerName
*/
- ConsumerBuilder consumerName(String consumerName);
+ ConsumerBuilder<T> consumerName(String consumerName);
/**
* If enabled, the consumer will read messages from the compacted topic rather than reading the full message backlog
@@ -229,7 +229,7 @@ public interface ConsumerBuilder extends Serializable, Cloneable {
* @param readCompacted
* whether to read from the compacted topic
*/
- ConsumerBuilder readCompacted(boolean readCompacted);
+ ConsumerBuilder<T> readCompacted(boolean readCompacted);
/**
* Set topics auto discovery period when using a pattern for topics consumer.
@@ -260,7 +260,7 @@ public interface ConsumerBuilder extends Serializable, Cloneable {
*
* @param priorityLevel
*/
- ConsumerBuilder priorityLevel(int priorityLevel);
+ ConsumerBuilder<T> priorityLevel(int priorityLevel);
/**
* Set a name/value property with this consumer.
@@ -269,7 +269,7 @@ public interface ConsumerBuilder extends Serializable, Cloneable {
* @param value
* @return
*/
- ConsumerBuilder property(String key, String value);
+ ConsumerBuilder<T> property(String key, String value);
/**
* Add all the properties in the provided map
@@ -277,6 +277,6 @@ public interface ConsumerBuilder extends Serializable, Cloneable {
* @param properties
* @return
*/
- ConsumerBuilder properties(Map<String, String> properties);
+ ConsumerBuilder<T> properties(Map<String, String> properties);
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerConfiguration.java b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerConfiguration.java
index 1c68c32..7761b9d 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerConfiguration.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerConfiguration.java
@@ -44,7 +44,7 @@ public class ConsumerConfiguration implements Serializable {
private static final long serialVersionUID = 1L;
- private final ConsumerConfigurationData conf = new ConsumerConfigurationData();
+ private final ConsumerConfigurationData<byte[]> conf = new ConsumerConfigurationData<>();
/**
* @return the configured timeout in milliseconds for unacked messages.
@@ -193,7 +193,7 @@ public class ConsumerConfiguration implements Serializable {
/**
* Sets the ConsumerCryptoFailureAction to the value specified
*
- * @param The
+ * @param action
* consumer action
*/
public void setCryptoFailureAction(ConsumerCryptoFailureAction action) {
@@ -336,7 +336,7 @@ public class ConsumerConfiguration implements Serializable {
return conf.getProperties();
}
- public ConsumerConfigurationData getConfigurationData() {
+ public ConsumerConfigurationData<byte[]> getConfigurationData() {
return conf;
}
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/api/Message.java b/pulsar-client/src/main/java/org/apache/pulsar/client/api/Message.java
index 27e0ac2..33be458 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/api/Message.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/api/Message.java
@@ -1,3 +1,4 @@
+
/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
@@ -25,7 +26,7 @@ import java.util.Map;
*
*
*/
-public interface Message {
+public interface Message<T> {
/**
* Return the properties attached to the message.
@@ -62,6 +63,8 @@ public interface Message {
*/
byte[] getData();
+ T getValue();
+
/**
* Get the unique message ID associated with this message.
*
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/api/MessageBuilder.java b/pulsar-client/src/main/java/org/apache/pulsar/client/api/MessageBuilder.java
index e4baf06..d6a013b 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/api/MessageBuilder.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/api/MessageBuilder.java
@@ -28,7 +28,7 @@ import org.apache.pulsar.client.impl.MessageBuilderImpl;
* Message builder factory. Use this class to create messages to be send to the Pulsar producer
*
*/
-public interface MessageBuilder {
+public interface MessageBuilder<T> {
/**
* Create a new message builder instance.
* <p>
@@ -36,8 +36,12 @@ public interface MessageBuilder {
*
* @return a new message builder
*/
- public static MessageBuilder create() {
- return new MessageBuilderImpl();
+ static <T> MessageBuilder<T> create(Schema<T> schema) {
+ return new MessageBuilderImpl<>(schema);
+ }
+
+ static MessageBuilder<byte[]> create() {
+ return create(Schema.IDENTITY);
}
/**
@@ -45,7 +49,15 @@ public interface MessageBuilder {
*
* @return a {@link Message} ready to be sent through a {@link Producer}
*/
- Message build();
+ Message<T> build();
+
+ /**
+ * Set a domain object on the message
+ *
+ * @param value
+ * the domain object
+ */
+ MessageBuilder<T> setValue(T value);
/**
* Set the content of the message
@@ -53,7 +65,7 @@ public interface MessageBuilder {
* @param data
* array containing the payload
*/
- MessageBuilder setContent(byte[] data);
+ MessageBuilder<T> setContent(byte[] data);
/**
* Set the content of the message
@@ -65,7 +77,7 @@ public interface MessageBuilder {
* @param length
* length of the payload starting from the above offset
*/
- MessageBuilder setContent(byte[] data, int offet, int length);
+ MessageBuilder<T> setContent(byte[] data, int offset, int length);
/**
* Set the content of the message
@@ -73,7 +85,7 @@ public interface MessageBuilder {
* @param buf
* a {@link ByteBuffer} with the payload of the message
*/
- MessageBuilder setContent(ByteBuffer buf);
+ MessageBuilder<T> setContent(ByteBuffer buf);
/**
* Sets a new property on a message.
@@ -83,19 +95,19 @@ public interface MessageBuilder {
* @param value
* the associated value
*/
- MessageBuilder setProperty(String name, String value);
+ MessageBuilder<T> setProperty(String name, String value);
/**
* Add all the properties in the provided map
*/
- MessageBuilder setProperties(Map<String, String> properties);
+ MessageBuilder<T> setProperties(Map<String, String> properties);
/**
* Sets the key of the message for routing policy
*
* @param key
*/
- MessageBuilder setKey(String key);
+ MessageBuilder<T> setKey(String key);
/**
* Set the event time for a given message.
@@ -107,7 +119,7 @@ public interface MessageBuilder {
*
* @since 1.20.0
*/
- MessageBuilder setEventTime(long timestamp);
+ MessageBuilder<T> setEventTime(long timestamp);
/**
* Specify a custom sequence id for the message being published.
@@ -125,17 +137,17 @@ public interface MessageBuilder {
* the sequence id to assign to the current message
* @since 1.20.0
*/
- MessageBuilder setSequenceId(long sequenceId);
+ MessageBuilder<T> setSequenceId(long sequenceId);
/**
* Override the replication clusters for this message.
*
* @param clusters
*/
- MessageBuilder setReplicationClusters(List<String> clusters);
+ MessageBuilder<T> setReplicationClusters(List<String> clusters);
/**
* Disable replication for this message.
*/
- MessageBuilder disableReplication();
+ MessageBuilder<T> disableReplication();
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/api/MessageListener.java b/pulsar-client/src/main/java/org/apache/pulsar/client/api/MessageListener.java
index a6623f4..b0676e3 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/api/MessageListener.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/api/MessageListener.java
@@ -25,7 +25,7 @@ import java.io.Serializable;
*
*
*/
-public interface MessageListener extends Serializable {
+public interface MessageListener<T> extends Serializable {
/**
* This method is called whenever a new message is received.
*
@@ -40,7 +40,7 @@ public interface MessageListener extends Serializable {
* @param msg
* the message object
*/
- void received(Consumer consumer, Message msg);
+ void received(Consumer<T> consumer, Message<T> msg);
/**
* Get the notification when a topic is terminated
@@ -48,7 +48,7 @@ public interface MessageListener extends Serializable {
* @param consumer
* the Consumer object associated with the terminated topic
*/
- default void reachedEndOfTopic(Consumer consumer) {
+ default void reachedEndOfTopic(Consumer<T> consumer) {
// By default ignore the notification
}
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/api/Producer.java b/pulsar-client/src/main/java/org/apache/pulsar/client/api/Producer.java
index 878899c..773d36c 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/api/Producer.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/api/Producer.java
@@ -30,7 +30,7 @@ import org.apache.pulsar.client.impl.ProducerStats;
*
*
*/
-public interface Producer extends Closeable {
+public interface Producer<T> extends Closeable {
/**
* @return the topic which producer is publishing to
@@ -53,12 +53,12 @@ public interface Producer extends Closeable {
* @throws PulsarClientException.AlreadyClosedException
* if the producer was already closed
*/
- MessageId send(byte[] message) throws PulsarClientException;
+ MessageId send(T message) throws PulsarClientException;
/**
* Send a message asynchronously
* <p>
- * When the producer queue is full, by default this method will complete the future with an exception {@link PulsarClientException#ProducerQueueIsFullError}
+ * When the producer queue is full, by default this method will complete the future with an exception {@link PulsarClientException.ProducerQueueIsFullError}
* <p>
* See {@link ProducerConfiguration#setMaxPendingMessages} to configure the producer queue size and
* {@link ProducerConfiguration#setBlockIfQueueFull(boolean)} to change the blocking behavior.
@@ -67,7 +67,7 @@ public interface Producer extends Closeable {
* a byte array with the payload of the message
* @return a future that can be used to track when the message will have been safely persisted
*/
- CompletableFuture<MessageId> sendAsync(byte[] message);
+ CompletableFuture<MessageId> sendAsync(T message);
/**
* Send a message
@@ -78,12 +78,12 @@ public interface Producer extends Closeable {
* @throws PulsarClientException.TimeoutException
* if the message was not correctly received by the system within the timeout period
*/
- MessageId send(Message message) throws PulsarClientException;
+ MessageId send(Message<T> message) throws PulsarClientException;
/**
* Send a message asynchronously
* <p>
- * When the returned {@link CompletatableFuture} is marked as completed successfully, the provided message will
+ * When the returned {@link CompletableFuture} is marked as completed successfully, the provided message will
* contain the {@link MessageId} assigned by the broker to the published message.
* <p>
* Example:
@@ -97,7 +97,7 @@ public interface Producer extends Closeable {
* });</code>
* </pre>
* <p>
- * When the producer queue is full, by default this method will complete the future with an exception {@link PulsarClientException#ProducerQueueIsFullError}
+ * When the producer queue is full, by default this method will complete the future with an exception {@link PulsarClientException.ProducerQueueIsFullError}
* <p>
* See {@link ProducerConfiguration#setMaxPendingMessages} to configure the producer queue size and
* {@link ProducerConfiguration#setBlockIfQueueFull(boolean)} to change the blocking behavior.
@@ -106,7 +106,7 @@ public interface Producer extends Closeable {
* a message
* @return a future that can be used to track when the message will have been safely persisted
*/
- CompletableFuture<MessageId> sendAsync(Message message);
+ CompletableFuture<MessageId> sendAsync(Message<T> message);
/**
* Get the last sequence id that was published by this producer.
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java
index 3961f8c..a037cfe 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java
@@ -30,7 +30,7 @@ import org.apache.pulsar.client.api.PulsarClientException.ProducerQueueIsFullErr
*
* @see PulsarClient#newProducer()
*/
-public interface ProducerBuilder extends Serializable, Cloneable {
+public interface ProducerBuilder<T> extends Serializable, Cloneable {
/**
* Finalize the creation of the {@link Producer} instance.
@@ -43,7 +43,7 @@ public interface ProducerBuilder extends Serializable, Cloneable {
* @throws PulsarClientException
* if the producer creation fails
*/
- Producer create() throws PulsarClientException;
+ Producer<T> create() throws PulsarClientException;
/**
* Finalize the creation of the {@link Producer} instance in asynchronous mode.
@@ -56,7 +56,7 @@ public interface ProducerBuilder extends Serializable, Cloneable {
* @throws PulsarClientException
* if the producer creation fails
*/
- CompletableFuture<Producer> createAsync();
+ CompletableFuture<Producer<T>> createAsync();
/**
* Create a copy of the current {@link ProducerBuilder}.
@@ -71,7 +71,7 @@ public interface ProducerBuilder extends Serializable, Cloneable {
* Producer producer2 = builder.clone().topic(TOPIC_2).create();
* </pre>
*/
- ProducerBuilder clone();
+ ProducerBuilder<T> clone();
/**
* Specify the topic this producer will be publishing on.
@@ -80,7 +80,7 @@ public interface ProducerBuilder extends Serializable, Cloneable {
*
* @param topicName
*/
- ProducerBuilder topic(String topicName);
+ ProducerBuilder<T> topic(String topicName);
/**
* Specify a name for the producer
@@ -95,7 +95,7 @@ public interface ProducerBuilder extends Serializable, Cloneable {
* @param producerName
* the custom name to use for the producer
*/
- ProducerBuilder producerName(String producerName);
+ ProducerBuilder<T> producerName(String producerName);
/**
* Set the send timeout <i>(default: 30 seconds)</i>
@@ -107,7 +107,7 @@ public interface ProducerBuilder extends Serializable, Cloneable {
* @param unit
* the time unit of the {@code sendTimeout}
*/
- ProducerBuilder sendTimeout(int sendTimeout, TimeUnit unit);
+ ProducerBuilder<T> sendTimeout(int sendTimeout, TimeUnit unit);
/**
* Set the max size of the queue holding the messages pending to receive an acknowledgment from the broker.
@@ -118,7 +118,7 @@ public interface ProducerBuilder extends Serializable, Cloneable {
* @param maxPendingMessages
* @return
*/
- ProducerBuilder maxPendingMessages(int maxPendingMessages);
+ ProducerBuilder<T> maxPendingMessages(int maxPendingMessages);
/**
* Set the number of max pending messages across all the partitions
@@ -128,7 +128,7 @@ public interface ProducerBuilder extends Serializable, Cloneable {
*
* @param maxPendingMessagesAcrossPartitions
*/
- ProducerBuilder maxPendingMessagesAcrossPartitions(int maxPendingMessagesAcrossPartitions);
+ ProducerBuilder<T> maxPendingMessagesAcrossPartitions(int maxPendingMessagesAcrossPartitions);
/**
* Set whether the {@link Producer#send} and {@link Producer#sendAsync} operations should block when the outgoing
@@ -141,7 +141,7 @@ public interface ProducerBuilder extends Serializable, Cloneable {
* whether to block {@link Producer#send} and {@link Producer#sendAsync} operations on queue full
* @return
*/
- ProducerBuilder blockIfQueueFull(boolean blockIfQueueFull);
+ ProducerBuilder<T> blockIfQueueFull(boolean blockIfQueueFull);
/**
* Set the message routing mode for the partitioned producer.
@@ -154,7 +154,7 @@ public interface ProducerBuilder extends Serializable, Cloneable {
* @param messageRoutingMode
* the message routing mode
*/
- ProducerBuilder messageRoutingMode(MessageRoutingMode messageRoutingMode);
+ ProducerBuilder<T> messageRoutingMode(MessageRoutingMode messageRoutingMode);
/**
* Change the {@link HashingScheme} used to chose the partition on where to publish a particular message.
@@ -171,7 +171,7 @@ public interface ProducerBuilder extends Serializable, Cloneable {
* @param hashingScheme
* the chosen {@link HashingScheme}
*/
- ProducerBuilder hashingScheme(HashingScheme hashingScheme);
+ ProducerBuilder<T> hashingScheme(HashingScheme hashingScheme);
/**
* Set the compression type for the producer.
@@ -185,7 +185,7 @@ public interface ProducerBuilder extends Serializable, Cloneable {
* @param compressionType
* @return
*/
- ProducerBuilder compressionType(CompressionType compressionType);
+ ProducerBuilder<T> compressionType(CompressionType compressionType);
/**
* Set a custom message routing policy by passing an implementation of MessageRouter
@@ -193,7 +193,7 @@ public interface ProducerBuilder extends Serializable, Cloneable {
*
* @param messageRouter
*/
- ProducerBuilder messageRouter(MessageRouter messageRouter);
+ ProducerBuilder<T> messageRouter(MessageRouter messageRouter);
/**
* Control whether automatic batching of messages is enabled for the producer. <i>default: false [No batching]</i>
@@ -207,7 +207,7 @@ public interface ProducerBuilder extends Serializable, Cloneable {
*
* @see #batchingMaxPublishDelay(long, TimeUnit)
*/
- ProducerBuilder enableBatching(boolean enableBatching);
+ ProducerBuilder<T> enableBatching(boolean enableBatching);
/**
* Sets a {@link CryptoKeyReader}
@@ -215,7 +215,7 @@ public interface ProducerBuilder extends Serializable, Cloneable {
* @param cryptoKeyReader
* CryptoKeyReader object
*/
- ProducerBuilder cryptoKeyReader(CryptoKeyReader cryptoKeyReader);
+ ProducerBuilder<T> cryptoKeyReader(CryptoKeyReader cryptoKeyReader);
/**
* Add public encryption key, used by producer to encrypt the data key.
@@ -226,21 +226,21 @@ public interface ProducerBuilder extends Serializable, Cloneable {
* after compression. If batch messaging is enabled, the batched message is encrypted.
*
*/
- ProducerBuilder addEncryptionKey(String key);
+ ProducerBuilder<T> addEncryptionKey(String key);
/**
* Sets the ProducerCryptoFailureAction to the value specified
*
- * @param The
+ * @param action
* producer action
*/
- ProducerBuilder cryptoFailureAction(ProducerCryptoFailureAction action);
+ ProducerBuilder<T> cryptoFailureAction(ProducerCryptoFailureAction action);
/**
* Set the time period within which the messages sent will be batched <i>default: 10ms</i> if batch messages are
* enabled. If set to a non zero value, messages will be queued until this time interval or until
*
- * @see ProducerConfiguration#batchingMaxMessages threshold is reached; all messages will be published as a single
+ * @see ProducerConfiguration#getBatchingMaxMessages() threshold is reached; all messages will be published as a single
* batch message. The consumer will be delivered individual messages in the batch in the same order they were
* enqueued
* @param batchDelay
@@ -249,7 +249,7 @@ public interface ProducerBuilder extends Serializable, Cloneable {
* the time unit of the {@code batchDelay}
* @return
*/
- ProducerBuilder batchingMaxPublishDelay(long batchDelay, TimeUnit timeUnit);
+ ProducerBuilder<T> batchingMaxPublishDelay(long batchDelay, TimeUnit timeUnit);
/**
* Set the maximum number of messages permitted in a batch. <i>default: 1000</i> If set to a value greater than 1,
@@ -273,7 +273,7 @@ public interface ProducerBuilder extends Serializable, Cloneable {
* @param initialSequenceId
* @return
*/
- ProducerBuilder initialSequenceId(long initialSequenceId);
+ ProducerBuilder<T> initialSequenceId(long initialSequenceId);
/**
* Set a name/value property with this producer.
@@ -282,7 +282,7 @@ public interface ProducerBuilder extends Serializable, Cloneable {
* @param value
* @return
*/
- ProducerBuilder property(String key, String value);
+ ProducerBuilder<T> property(String key, String value);
/**
* Add all the properties in the provided map
@@ -290,5 +290,5 @@ public interface ProducerBuilder extends Serializable, Cloneable {
* @param properties
* @return
*/
- ProducerBuilder properties(Map<String, String> properties);
+ ProducerBuilder<T> properties(Map<String, String> properties);
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/api/PulsarClient.java b/pulsar-client/src/main/java/org/apache/pulsar/client/api/PulsarClient.java
index 6ba2518..98af0b3 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/api/PulsarClient.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/api/PulsarClient.java
@@ -89,7 +89,25 @@ public interface PulsarClient extends Closeable {
*
* @since 2.0.0
*/
- ProducerBuilder newProducer();
+ ProducerBuilder<byte[]> newProducer();
+
+ /**
+ * Create a producer with default for publishing on a specific topic
+ * <p>
+ * Example:
+ *
+ * <code>
+ * Producer producer = client.newProducer(mySchema).topic(myTopic).create();
+ * </code>
+ *
+ * @param schema
+ * provide a way to convert between serialized data and domain objects
+ *
+ * @return a {@link ProducerBuilder} object to configure and construct the {@link Producer} instance
+ *
+ * @since 2.0.0
+ */
+ <T> ProducerBuilder<T> newProducer(Schema<T> schema);
/**
* Create a producer with default for publishing on a specific topic
@@ -98,7 +116,19 @@ public interface PulsarClient extends Closeable {
*
* @since 2.0.0
*/
- ConsumerBuilder newConsumer();
+ ConsumerBuilder<byte[]> newConsumer();
+
+ /**
+ * Create a producer with default for publishing on a specific topic
+ *
+ * @param schema
+ * provide a way to convert between serialized data and domain objects
+ *
+ * @return a {@link ProducerBuilder} object to configure and construct the {@link Producer} instance
+ *
+ * @since 2.0.0
+ */
+ <T> ConsumerBuilder<T> newConsumer(Schema<T> schema);
/**
* Create a topic reader for reading messages from the specified topic.
@@ -110,7 +140,22 @@ public interface PulsarClient extends Closeable {
*
* @since 2.0.0
*/
- ReaderBuilder newReader();
+ ReaderBuilder<byte[]> newReader();
+
+ /**
+ * Create a topic reader for reading messages from the specified topic.
+ * <p>
+ * The Reader provides a low-level abstraction that allows for manual positioning in the topic, without using a
+ * subscription. Reader can only work on non-partitioned topics.
+ *
+ * @param schema
+ * provide a way to convert between serialized data and domain objects
+ *
+ * @return a {@link ReaderBuilder} that can be used to configure and construct a {@link Reader} instance
+ *
+ * @since 2.0.0
+ */
+ <T> ReaderBuilder<T> newReader(Schema<T> schema);
/**
* Create a producer with default {@link ProducerConfiguration} for publishing on a specific topic
@@ -129,7 +174,7 @@ public interface PulsarClient extends Closeable {
* @deprecated use {@link #newProducer()} to build a new producer
*/
@Deprecated
- Producer createProducer(String topic) throws PulsarClientException;
+ Producer<byte[]> createProducer(String topic) throws PulsarClientException;
/**
* Asynchronously create a producer with default {@link ProducerConfiguration} for publishing on a specific topic
@@ -140,7 +185,7 @@ public interface PulsarClient extends Closeable {
* @deprecated use {@link #newProducer()} to build a new producer
*/
@Deprecated
- CompletableFuture<Producer> createProducerAsync(String topic);
+ CompletableFuture<Producer<byte[]>> createProducerAsync(String topic);
/**
* Create a producer with given {@code ProducerConfiguration} for publishing on a specific topic
@@ -156,7 +201,7 @@ public interface PulsarClient extends Closeable {
* @deprecated use {@link #newProducer()} to build a new producer
*/
@Deprecated
- Producer createProducer(String topic, ProducerConfiguration conf) throws PulsarClientException;
+ Producer<byte[]> createProducer(String topic, ProducerConfiguration conf) throws PulsarClientException;
/**
* Asynchronously create a producer with given {@code ProducerConfiguration} for publishing on a specific topic
@@ -169,7 +214,7 @@ public interface PulsarClient extends Closeable {
* @deprecated use {@link #newProducer()} to build a new producer
*/
@Deprecated
- CompletableFuture<Producer> createProducerAsync(String topic, ProducerConfiguration conf);
+ CompletableFuture<Producer<byte[]>> createProducerAsync(String topic, ProducerConfiguration conf);
/**
* Subscribe to the given topic and subscription combination with default {@code ConsumerConfiguration}
@@ -185,7 +230,7 @@ public interface PulsarClient extends Closeable {
* @deprecated Use {@link #newConsumer()} to build a new consumer
*/
@Deprecated
- Consumer subscribe(String topic, String subscription) throws PulsarClientException;
+ Consumer<byte[]> subscribe(String topic, String subscription) throws PulsarClientException;
/**
* Asynchronously subscribe to the given topic and subscription combination using default
@@ -199,7 +244,7 @@ public interface PulsarClient extends Closeable {
* @deprecated Use {@link #newConsumer()} to build a new consumer
*/
@Deprecated
- CompletableFuture<Consumer> subscribeAsync(String topic, String subscription);
+ CompletableFuture<Consumer<byte[]>> subscribeAsync(String topic, String subscription);
/**
* Subscribe to the given topic and subscription combination with given {@code ConsumerConfiguration}
@@ -215,7 +260,7 @@ public interface PulsarClient extends Closeable {
* @deprecated Use {@link #newConsumer()} to build a new consumer
*/
@Deprecated
- Consumer subscribe(String topic, String subscription, ConsumerConfiguration conf) throws PulsarClientException;
+ Consumer<byte[]> subscribe(String topic, String subscription, ConsumerConfiguration conf) throws PulsarClientException;
/**
* Asynchronously subscribe to the given topic and subscription combination using given
@@ -231,7 +276,7 @@ public interface PulsarClient extends Closeable {
* @deprecated Use {@link #newConsumer()} to build a new consumer
*/
@Deprecated
- CompletableFuture<Consumer> subscribeAsync(String topic, String subscription, ConsumerConfiguration conf);
+ CompletableFuture<Consumer<byte[]>> subscribeAsync(String topic, String subscription, ConsumerConfiguration conf);
/**
* Create a topic reader with given {@code ReaderConfiguration} for reading messages from the specified topic.
@@ -259,7 +304,7 @@ public interface PulsarClient extends Closeable {
* @deprecated Use {@link #newReader()} to build a new reader
*/
@Deprecated
- Reader createReader(String topic, MessageId startMessageId, ReaderConfiguration conf) throws PulsarClientException;
+ Reader<byte[]> createReader(String topic, MessageId startMessageId, ReaderConfiguration conf) throws PulsarClientException;
/**
* Asynchronously create a topic reader with given {@code ReaderConfiguration} for reading messages from the
@@ -288,7 +333,7 @@ public interface PulsarClient extends Closeable {
* @deprecated Use {@link #newReader()} to build a new reader
*/
@Deprecated
- CompletableFuture<Reader> createReaderAsync(String topic, MessageId startMessageId, ReaderConfiguration conf);
+ CompletableFuture<Reader<byte[]>> createReaderAsync(String topic, MessageId startMessageId, ReaderConfiguration conf);
/**
* Close the PulsarClient and release all the resources.
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/api/Reader.java b/pulsar-client/src/main/java/org/apache/pulsar/client/api/Reader.java
index d29b238..f343c90 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/api/Reader.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/api/Reader.java
@@ -26,7 +26,7 @@ import java.util.concurrent.TimeUnit;
* A Reader can be used to scan through all the messages currently available in a topic.
*
*/
-public interface Reader extends Closeable {
+public interface Reader<T> extends Closeable {
/**
* @return the topic from which this reader is reading from
@@ -39,7 +39,7 @@ public interface Reader extends Closeable {
* @return the next messasge
* @throws PulsarClientException
*/
- Message readNext() throws PulsarClientException;
+ Message<T> readNext() throws PulsarClientException;
/**
* Read the next message in the topic.
@@ -47,9 +47,9 @@ public interface Reader extends Closeable {
* @return the next messasge
* @throws PulsarClientException
*/
- Message readNext(int timeout, TimeUnit unit) throws PulsarClientException;
+ Message<T> readNext(int timeout, TimeUnit unit) throws PulsarClientException;
- CompletableFuture<Message> readNextAsync();
+ CompletableFuture<Message<T>> readNextAsync();
/**
* Asynchronously close the reader and stop the broker to push more messages
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderBuilder.java b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderBuilder.java
index 8c37899..c81b2eb 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderBuilder.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderBuilder.java
@@ -28,7 +28,7 @@ import java.util.concurrent.CompletableFuture;
*
* @since 2.0.0
*/
-public interface ReaderBuilder extends Serializable, Cloneable {
+public interface ReaderBuilder<T> extends Serializable, Cloneable {
/**
* Finalize the creation of the {@link Reader} instance.
@@ -40,7 +40,7 @@ public interface ReaderBuilder extends Serializable, Cloneable {
* @throws PulsarClientException
* if the reader creation fails
*/
- Reader create() throws PulsarClientException;
+ Reader<T> create() throws PulsarClientException;
/**
* Finalize the creation of the {@link Reader} instance in asynchronous mode.
@@ -52,7 +52,7 @@ public interface ReaderBuilder extends Serializable, Cloneable {
* @throws PulsarClientException
* if the reader creation fails
*/
- CompletableFuture<Reader> createAsync();
+ CompletableFuture<Reader<T>> createAsync();
/**
* Create a copy of the current {@link ReaderBuilder}.
@@ -67,7 +67,7 @@ public interface ReaderBuilder extends Serializable, Cloneable {
* Reader reader2 = builder.clone().topic(TOPIC_2).create();
* </pre>
*/
- ReaderBuilder clone();
+ ReaderBuilder<T> clone();
/**
* Specify the topic this consumer will subscribe on.
@@ -76,7 +76,7 @@ public interface ReaderBuilder extends Serializable, Cloneable {
*
* @param topicName
*/
- ReaderBuilder topic(String topicName);
+ ReaderBuilder<T> topic(String topicName);
/**
* The initial reader positioning is done by specifying a message id. The options are:
@@ -88,7 +88,7 @@ public interface ReaderBuilder extends Serializable, Cloneable {
* specific position. The first message to be read will be the message next to the specified messageId.
* </ul>
*/
- ReaderBuilder startMessageId(MessageId startMessageId);
+ ReaderBuilder<T> startMessageId(MessageId startMessageId);
/**
* Sets a {@link ReaderListener} for the reader
@@ -99,7 +99,7 @@ public interface ReaderBuilder extends Serializable, Cloneable {
* @param readerListener
* the listener object
*/
- ReaderBuilder readerListener(ReaderListener readerListener);
+ ReaderBuilder<T> readerListener(ReaderListener readerListener);
/**
* Sets a {@link CryptoKeyReader}
@@ -107,7 +107,7 @@ public interface ReaderBuilder extends Serializable, Cloneable {
* @param cryptoKeyReader
* CryptoKeyReader object
*/
- ReaderBuilder cryptoKeyReader(CryptoKeyReader cryptoKeyReader);
+ ReaderBuilder<T> cryptoKeyReader(CryptoKeyReader cryptoKeyReader);
/**
* Sets the ConsumerCryptoFailureAction to the value specified
@@ -115,7 +115,7 @@ public interface ReaderBuilder extends Serializable, Cloneable {
* @param action
* The action to take when the decoding fails
*/
- ReaderBuilder cryptoFailureAction(ConsumerCryptoFailureAction action);
+ ReaderBuilder<T> cryptoFailureAction(ConsumerCryptoFailureAction action);
/**
* Sets the size of the consumer receive queue.
@@ -129,19 +129,19 @@ public interface ReaderBuilder extends Serializable, Cloneable {
* @param receiverQueueSize
* the new receiver queue size value
*/
- ReaderBuilder receiverQueueSize(int receiverQueueSize);
+ ReaderBuilder<T> receiverQueueSize(int receiverQueueSize);
/**
* Set the reader name.
*
* @param readerName
*/
- ReaderBuilder readerName(String readerName);
+ ReaderBuilder<T> readerName(String readerName);
/**
* Set the subscription role prefix. The default prefix is "reader".
*
* @param subscriptionRolePrefix
*/
- ReaderBuilder subscriptionRolePrefix(String subscriptionRolePrefix);
+ ReaderBuilder<T> subscriptionRolePrefix(String subscriptionRolePrefix);
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderConfiguration.java b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderConfiguration.java
index bca04e5..243166c 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderConfiguration.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderConfiguration.java
@@ -34,12 +34,12 @@ import org.apache.pulsar.client.impl.conf.ReaderConfigurationData;
@Deprecated
public class ReaderConfiguration implements Serializable {
- private final ReaderConfigurationData conf = new ReaderConfigurationData();
+ private final ReaderConfigurationData<byte[]> conf = new ReaderConfigurationData<>();
/**
* @return the configured {@link ReaderListener} for the reader
*/
- public ReaderListener getReaderListener() {
+ public ReaderListener<byte[]> getReaderListener() {
return conf.getReaderListener();
}
@@ -52,7 +52,7 @@ public class ReaderConfiguration implements Serializable {
* @param readerListener
* the listener object
*/
- public ReaderConfiguration setReaderListener(ReaderListener readerListener) {
+ public ReaderConfiguration setReaderListener(ReaderListener<byte[]> readerListener) {
checkNotNull(readerListener);
conf.setReaderListener(readerListener);
return this;
@@ -86,7 +86,7 @@ public class ReaderConfiguration implements Serializable {
/**
* Sets the ConsumerCryptoFailureAction to the value specified
- *
+ *
* @param action
* The action to take when the decoding fails
*/
@@ -155,7 +155,7 @@ public class ReaderConfiguration implements Serializable {
return this;
}
- public ReaderConfigurationData getReaderConfigurationData() {
+ public ReaderConfigurationData<byte[]> getReaderConfigurationData() {
return conf;
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderListener.java b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderListener.java
index f295649..3d2af2a 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderListener.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/api/ReaderListener.java
@@ -23,7 +23,7 @@ import java.io.Serializable;
/**
* A listener that will be called in order for every message received.
*/
-public interface ReaderListener extends Serializable {
+public interface ReaderListener<T> extends Serializable {
/**
* This method is called whenever a new message is received.
*
@@ -38,7 +38,7 @@ public interface ReaderListener extends Serializable {
* @param msg
* the message object
*/
- void received(Reader reader, Message msg);
+ void received(Reader<T> reader, Message<T> msg);
/**
* Get the notification when a topic is terminated
@@ -46,7 +46,7 @@ public interface ReaderListener extends Serializable {
* @param reader
* the Reader object associated with the terminated topic
*/
- default void reachedEndOfTopic(Reader reader) {
+ default void reachedEndOfTopic(Reader<T> reader) {
// By default ignore the notification
}
}
diff --git a/pulsar-storm/src/main/java/org/apache/pulsar/storm/TupleToMessageMapper.java b/pulsar-client/src/main/java/org/apache/pulsar/client/api/Schema.java
similarity index 57%
copy from pulsar-storm/src/main/java/org/apache/pulsar/storm/TupleToMessageMapper.java
copy to pulsar-client/src/main/java/org/apache/pulsar/client/api/Schema.java
index 5c464d9..346c525 100644
--- a/pulsar-storm/src/main/java/org/apache/pulsar/storm/TupleToMessageMapper.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/api/Schema.java
@@ -16,29 +16,21 @@
* specific language governing permissions and limitations
* under the License.
*/
-package org.apache.pulsar.storm;
+package org.apache.pulsar.client.api;
-import java.io.Serializable;
+public interface Schema<T> {
+ byte[] encode(T message);
+ T decode(byte[] bytes);
-import org.apache.pulsar.client.api.Message;
+ Schema<byte[]> IDENTITY = new Schema<byte[]>() {
+ @Override
+ public byte[] encode(byte[] message) {
+ return message;
+ }
-import org.apache.storm.topology.OutputFieldsDeclarer;
-import org.apache.storm.tuple.Tuple;
-
-public interface TupleToMessageMapper extends Serializable {
-
- /**
- * Convert tuple to {@link org.apache.pulsar.client.api.Message}.
- *
- * @param tuple
- * @return
- */
- public Message toMessage(Tuple tuple);
-
- /**
- * Declare the output schema for the bolt.
- *
- * @param declarer
- */
- public void declareOutputFields(OutputFieldsDeclarer declarer);
+ @Override
+ public byte[] decode(byte[] bytes) {
+ return bytes;
+ }
+ };
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
index 842fbfb..af6b732 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBase.java
@@ -18,6 +18,7 @@
*/
package org.apache.pulsar.client.impl;
+import com.google.common.collect.Queues;
import java.util.Collections;
import java.util.Map;
import java.util.Set;
@@ -27,13 +28,13 @@ import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
-
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.ConsumerEventListener;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.MessageListener;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.apache.pulsar.client.util.ConsumerName;
@@ -42,9 +43,7 @@ import org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.SubType;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.collections.GrowableArrayBlockingQueue;
-import com.google.common.collect.Queues;
-
-public abstract class ConsumerBase extends HandlerBase implements Consumer {
+public abstract class ConsumerBase<T> extends HandlerBase implements Consumer<T> {
enum ConsumerType {
PARTITIONED, NON_PARTITIONED
@@ -53,16 +52,17 @@ public abstract class ConsumerBase extends HandlerBase implements Consumer {
protected final String subscription;
protected final ConsumerConfigurationData conf;
protected final String consumerName;
- protected final CompletableFuture<Consumer> subscribeFuture;
- protected final MessageListener listener;
+ protected final CompletableFuture<Consumer<T>> subscribeFuture;
+ protected final MessageListener<T> listener;
protected final ConsumerEventListener consumerEventListener;
protected final ExecutorService listenerExecutor;
- final BlockingQueue<Message> incomingMessages;
- protected final ConcurrentLinkedQueue<CompletableFuture<Message>> pendingReceives;
+ final BlockingQueue<Message<T>> incomingMessages;
+ protected final ConcurrentLinkedQueue<CompletableFuture<Message<T>>> pendingReceives;
protected int maxReceiverQueueSize;
+ protected Schema<T> schema;
- protected ConsumerBase(PulsarClientImpl client, String topic, ConsumerConfigurationData conf, int receiverQueueSize,
- ExecutorService listenerExecutor, CompletableFuture<Consumer> subscribeFuture) {
+ protected ConsumerBase(PulsarClientImpl client, String topic, ConsumerConfigurationData<T> conf, int receiverQueueSize,
+ ExecutorService listenerExecutor, CompletableFuture<Consumer<T>> subscribeFuture, Schema<T> schema) {
super(client, topic, new Backoff(100, TimeUnit.MILLISECONDS, 60, TimeUnit.SECONDS, 0, TimeUnit.MILLISECONDS));
this.maxReceiverQueueSize = receiverQueueSize;
this.subscription = conf.getSubscriptionName();
@@ -79,10 +79,11 @@ public abstract class ConsumerBase extends HandlerBase implements Consumer {
this.listenerExecutor = listenerExecutor;
this.pendingReceives = Queues.newConcurrentLinkedQueue();
+ this.schema = schema;
}
@Override
- public Message receive() throws PulsarClientException {
+ public Message<T> receive() throws PulsarClientException {
if (listener != null) {
throw new PulsarClientException.InvalidConfigurationException(
"Cannot use receive() when a listener has been set");
@@ -108,7 +109,7 @@ public abstract class ConsumerBase extends HandlerBase implements Consumer {
}
@Override
- public CompletableFuture<Message> receiveAsync() {
+ public CompletableFuture<Message<T>> receiveAsync() {
if (listener != null) {
return FutureUtil.failedFuture(new PulsarClientException.InvalidConfigurationException(
@@ -132,12 +133,12 @@ public abstract class ConsumerBase extends HandlerBase implements Consumer {
return internalReceiveAsync();
}
- abstract protected Message internalReceive() throws PulsarClientException;
+ abstract protected Message<T> internalReceive() throws PulsarClientException;
- abstract protected CompletableFuture<Message> internalReceiveAsync();
+ abstract protected CompletableFuture<Message<T>> internalReceiveAsync();
@Override
- public Message receive(int timeout, TimeUnit unit) throws PulsarClientException {
+ public Message<T> receive(int timeout, TimeUnit unit) throws PulsarClientException {
if (conf.getReceiverQueueSize() == 0) {
throw new PulsarClientException.InvalidConfigurationException(
"Can't use receive with timeout, if the queue size is 0");
@@ -164,7 +165,7 @@ public abstract class ConsumerBase extends HandlerBase implements Consumer {
return internalReceive(timeout, unit);
}
- abstract protected Message internalReceive(int timeout, TimeUnit unit) throws PulsarClientException;
+ abstract protected Message<T> internalReceive(int timeout, TimeUnit unit) throws PulsarClientException;
@Override
public void acknowledge(Message message) throws PulsarClientException {
@@ -321,7 +322,7 @@ public abstract class ConsumerBase extends HandlerBase implements Consumer {
abstract public int numMessagesInQueue();
- public CompletableFuture<Consumer> subscribeFuture() {
+ public CompletableFuture<Consumer<T>> subscribeFuture() {
return subscribeFuture;
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java
index 4c9caa1..e7163c9 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerBuilderImpl.java
@@ -33,37 +33,40 @@ import org.apache.pulsar.client.api.ConsumerCryptoFailureAction;
import org.apache.pulsar.client.api.CryptoKeyReader;
import org.apache.pulsar.client.api.MessageListener;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.apache.pulsar.common.util.FutureUtil;
import com.google.common.collect.Lists;
-public class ConsumerBuilderImpl implements ConsumerBuilder {
+public class ConsumerBuilderImpl<T> implements ConsumerBuilder<T> {
private static final long serialVersionUID = 1L;
private final PulsarClientImpl client;
- private final ConsumerConfigurationData conf;
+ private final ConsumerConfigurationData<T> conf;
+ private final Schema<T> schema;
private static long MIN_ACK_TIMEOUT_MILLIS = 1000;
- ConsumerBuilderImpl(PulsarClientImpl client) {
- this(client, new ConsumerConfigurationData());
+ ConsumerBuilderImpl(PulsarClientImpl client, Schema<T> schema) {
+ this(client, new ConsumerConfigurationData<T>(), schema);
}
- private ConsumerBuilderImpl(PulsarClientImpl client, ConsumerConfigurationData conf) {
+ private ConsumerBuilderImpl(PulsarClientImpl client, ConsumerConfigurationData<T> conf, Schema<T> schema) {
this.client = client;
this.conf = conf;
+ this.schema = schema;
}
@Override
- public ConsumerBuilder clone() {
- return new ConsumerBuilderImpl(client, conf.clone());
+ public ConsumerBuilder<T> clone() {
+ return new ConsumerBuilderImpl<>(client, conf.clone(), schema);
}
@Override
- public Consumer subscribe() throws PulsarClientException {
+ public Consumer<T> subscribe() throws PulsarClientException {
try {
return subscribeAsync().get();
} catch (ExecutionException e) {
@@ -80,7 +83,7 @@ public class ConsumerBuilderImpl implements ConsumerBuilder {
}
@Override
- public CompletableFuture<Consumer> subscribeAsync() {
+ public CompletableFuture<Consumer<T>> subscribeAsync() {
if (conf.getTopicNames().isEmpty() && conf.getTopicsPattern() == null) {
return FutureUtil
.failedFuture(new IllegalArgumentException("Topic name must be set on the consumer builder"));
@@ -91,45 +94,45 @@ public class ConsumerBuilderImpl implements ConsumerBuilder {
new IllegalArgumentException("Subscription name must be set on the consumer builder"));
}
- return client.subscribeAsync(conf);
+ return client.subscribeAsync(conf, schema);
}
@Override
- public ConsumerBuilder topic(String... topicNames) {
+ public ConsumerBuilder<T> topic(String... topicNames) {
checkArgument(topicNames.length > 0, "Passed in topicNames should not be empty.");
conf.getTopicNames().addAll(Lists.newArrayList(topicNames));
return this;
}
@Override
- public ConsumerBuilder topics(List<String> topicNames) {
+ public ConsumerBuilder<T> topics(List<String> topicNames) {
checkArgument(topicNames != null && !topicNames.isEmpty(), "Passed in topicNames list should not be empty.");
conf.getTopicNames().addAll(topicNames);
return this;
}
@Override
- public ConsumerBuilder topicsPattern(Pattern topicsPattern) {
+ public ConsumerBuilder<T> topicsPattern(Pattern topicsPattern) {
checkArgument(conf.getTopicsPattern() == null, "Pattern has already been set.");
conf.setTopicsPattern(topicsPattern);
return this;
}
@Override
- public ConsumerBuilder topicsPattern(String topicsPattern) {
+ public ConsumerBuilder<T> topicsPattern(String topicsPattern) {
checkArgument(conf.getTopicsPattern() == null, "Pattern has already been set.");
conf.setTopicsPattern(Pattern.compile(topicsPattern));
return this;
}
@Override
- public ConsumerBuilder subscriptionName(String subscriptionName) {
+ public ConsumerBuilder<T> subscriptionName(String subscriptionName) {
conf.setSubscriptionName(subscriptionName);
return this;
}
@Override
- public ConsumerBuilder ackTimeout(long ackTimeout, TimeUnit timeUnit) {
+ public ConsumerBuilder<T> ackTimeout(long ackTimeout, TimeUnit timeUnit) {
checkArgument(timeUnit.toMillis(ackTimeout) >= MIN_ACK_TIMEOUT_MILLIS,
"Ack timeout should be should be greater than " + MIN_ACK_TIMEOUT_MILLIS + " ms");
conf.setAckTimeoutMillis(timeUnit.toMillis(ackTimeout));
@@ -137,67 +140,67 @@ public class ConsumerBuilderImpl implements ConsumerBuilder {
}
@Override
- public ConsumerBuilder subscriptionType(SubscriptionType subscriptionType) {
+ public ConsumerBuilder<T> subscriptionType(SubscriptionType subscriptionType) {
conf.setSubscriptionType(subscriptionType);
return this;
}
@Override
- public ConsumerBuilder messageListener(MessageListener messageListener) {
+ public ConsumerBuilder<T> messageListener(MessageListener<T> messageListener) {
conf.setMessageListener(messageListener);
return this;
}
@Override
- public ConsumerBuilder cryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
+ public ConsumerBuilder<T> cryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
conf.setCryptoKeyReader(cryptoKeyReader);
return this;
}
@Override
- public ConsumerBuilder cryptoFailureAction(ConsumerCryptoFailureAction action) {
+ public ConsumerBuilder<T> cryptoFailureAction(ConsumerCryptoFailureAction action) {
conf.setCryptoFailureAction(action);
return this;
}
@Override
- public ConsumerBuilder receiverQueueSize(int receiverQueueSize) {
+ public ConsumerBuilder<T> receiverQueueSize(int receiverQueueSize) {
conf.setReceiverQueueSize(receiverQueueSize);
return this;
}
@Override
- public ConsumerBuilder consumerName(String consumerName) {
+ public ConsumerBuilder<T> consumerName(String consumerName) {
conf.setConsumerName(consumerName);
return this;
}
@Override
- public ConsumerBuilder priorityLevel(int priorityLevel) {
+ public ConsumerBuilder<T> priorityLevel(int priorityLevel) {
conf.setPriorityLevel(priorityLevel);
return this;
}
@Override
- public ConsumerBuilder property(String key, String value) {
+ public ConsumerBuilder<T> property(String key, String value) {
conf.getProperties().put(key, value);
return this;
}
@Override
- public ConsumerBuilder properties(Map<String, String> properties) {
+ public ConsumerBuilder<T> properties(Map<String, String> properties) {
conf.getProperties().putAll(properties);
return this;
}
@Override
- public ConsumerBuilder maxTotalReceiverQueueSizeAcrossPartitions(int maxTotalReceiverQueueSizeAcrossPartitions) {
+ public ConsumerBuilder<T> maxTotalReceiverQueueSizeAcrossPartitions(int maxTotalReceiverQueueSizeAcrossPartitions) {
conf.setMaxTotalReceiverQueueSizeAcrossPartitions(maxTotalReceiverQueueSizeAcrossPartitions);
return this;
}
@Override
- public ConsumerBuilder readCompacted(boolean readCompacted) {
+ public ConsumerBuilder<T> readCompacted(boolean readCompacted) {
conf.setReadCompacted(readCompacted);
return this;
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index 1119f4f..8499565 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -25,6 +25,11 @@ import static java.lang.String.format;
import static org.apache.pulsar.common.api.Commands.hasChecksum;
import static org.apache.pulsar.common.api.Commands.readChecksum;
+import com.google.common.collect.Iterables;
+import io.netty.buffer.ByteBuf;
+import io.netty.util.Timeout;
+import io.netty.util.concurrent.Future;
+import io.netty.util.concurrent.GenericFutureListener;
import java.io.IOException;
import java.util.ArrayList;
import java.util.BitSet;
@@ -47,12 +52,12 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Collectors;
-
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.ConsumerCryptoFailureAction;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.apache.pulsar.common.api.Commands;
@@ -70,14 +75,7 @@ import org.apache.pulsar.common.util.FutureUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.collect.Iterables;
-
-import io.netty.buffer.ByteBuf;
-import io.netty.util.Timeout;
-import io.netty.util.concurrent.Future;
-import io.netty.util.concurrent.GenericFutureListener;
-
-public class ConsumerImpl extends ConsumerBase {
+public class ConsumerImpl<T> extends ConsumerBase<T> {
private static final int MAX_REDELIVER_UNACKNOWLEDGED = 1000;
private final long consumerId;
@@ -129,15 +127,15 @@ public class ConsumerImpl extends ConsumerBase {
NonDurable
}
- ConsumerImpl(PulsarClientImpl client, String topic, ConsumerConfigurationData conf,
- ExecutorService listenerExecutor, int partitionIndex, CompletableFuture<Consumer> subscribeFuture) {
- this(client, topic, conf, listenerExecutor, partitionIndex, subscribeFuture, SubscriptionMode.Durable, null);
+ ConsumerImpl(PulsarClientImpl client, String topic, ConsumerConfigurationData<T> conf,
+ ExecutorService listenerExecutor, int partitionIndex, CompletableFuture<Consumer<T>> subscribeFuture, Schema<T> schema) {
+ this(client, topic, conf, listenerExecutor, partitionIndex, subscribeFuture, SubscriptionMode.Durable, null, schema);
}
- ConsumerImpl(PulsarClientImpl client, String topic, ConsumerConfigurationData conf,
- ExecutorService listenerExecutor, int partitionIndex, CompletableFuture<Consumer> subscribeFuture,
- SubscriptionMode subscriptionMode, MessageId startMessageId) {
- super(client, topic, conf, conf.getReceiverQueueSize(), listenerExecutor, subscribeFuture);
+ ConsumerImpl(PulsarClientImpl client, String topic, ConsumerConfigurationData<T> conf,
+ ExecutorService listenerExecutor, int partitionIndex, CompletableFuture<Consumer<T>> subscribeFuture,
+ SubscriptionMode subscriptionMode, MessageId startMessageId, Schema<T> schema) {
+ super(client, topic, conf, conf.getReceiverQueueSize(), listenerExecutor, subscribeFuture, schema);
this.consumerId = client.newConsumerId();
this.subscriptionMode = subscriptionMode;
this.startMessageId = startMessageId != null ? new BatchMessageIdImpl((MessageIdImpl) startMessageId) : null;
@@ -219,7 +217,7 @@ public class ConsumerImpl extends ConsumerBase {
}
@Override
- protected Message internalReceive() throws PulsarClientException {
+ protected Message<T> internalReceive() throws PulsarClientException {
if (conf.getReceiverQueueSize() == 0) {
checkArgument(zeroQueueLock != null, "Receiver queue size can't be modified");
zeroQueueLock.writeLock().lock();
@@ -229,7 +227,7 @@ public class ConsumerImpl extends ConsumerBase {
zeroQueueLock.writeLock().unlock();
}
}
- Message message;
+ Message<T> message;
try {
message = incomingMessages.take();
messageProcessed(message);
@@ -242,10 +240,10 @@ public class ConsumerImpl extends ConsumerBase {
}
@Override
- protected CompletableFuture<Message> internalReceiveAsync() {
+ protected CompletableFuture<Message<T>> internalReceiveAsync() {
- CompletableFuture<Message> result = new CompletableFuture<Message>();
- Message message = null;
+ CompletableFuture<Message<T>> result = new CompletableFuture<>();
+ Message<T> message = null;
try {
lock.writeLock().lock();
message = incomingMessages.poll(0, TimeUnit.MILLISECONDS);
@@ -269,7 +267,7 @@ public class ConsumerImpl extends ConsumerBase {
return result;
}
- private Message fetchSingleMessageFromBroker() throws PulsarClientException {
+ private Message<T> fetchSingleMessageFromBroker() throws PulsarClientException {
checkArgument(conf.getReceiverQueueSize() == 0);
// Just being cautious
@@ -278,7 +276,7 @@ public class ConsumerImpl extends ConsumerBase {
incomingMessages.clear();
}
- Message message;
+ Message<T> message;
try {
// is cnx is null or if the connection breaks the connectionOpened function will send the flow again
waitingOnReceiveForZeroQueueSize = true;
@@ -317,8 +315,8 @@ public class ConsumerImpl extends ConsumerBase {
}
@Override
- protected Message internalReceive(int timeout, TimeUnit unit) throws PulsarClientException {
- Message message;
+ protected Message<T> internalReceive(int timeout, TimeUnit unit) throws PulsarClientException {
+ Message<T> message;
try {
message = incomingMessages.poll(timeout, unit);
if (message != null) {
@@ -736,7 +734,7 @@ public class ConsumerImpl extends ConsumerBase {
try {
if (listenerExecutor != null && !listenerExecutor.isShutdown()) {
while (!pendingReceives.isEmpty()) {
- CompletableFuture<Message> receiveFuture = pendingReceives.poll();
+ CompletableFuture<Message<T>> receiveFuture = pendingReceives.poll();
if (receiveFuture != null) {
receiveFuture.completeExceptionally(
new PulsarClientException.AlreadyClosedException("Consumer is already closed"));
@@ -801,8 +799,8 @@ public class ConsumerImpl extends ConsumerBase {
final int numMessages = msgMetadata.getNumMessagesInBatch();
if (numMessages == 1 && !msgMetadata.hasNumMessagesInBatch()) {
- final MessageImpl message = new MessageImpl(messageId, msgMetadata, uncompressedPayload,
- getPartitionIndex(), cnx);
+ final MessageImpl<T> message = new MessageImpl<>(messageId, msgMetadata, uncompressedPayload,
+ getPartitionIndex(), cnx, schema);
uncompressedPayload.release();
msgMetadata.recycle();
@@ -850,7 +848,7 @@ public class ConsumerImpl extends ConsumerBase {
listenerExecutor.execute(() -> {
for (int i = 0; i < numMessages; i++) {
try {
- Message msg = internalReceive(0, TimeUnit.MILLISECONDS);
+ Message<T> msg = internalReceive(0, TimeUnit.MILLISECONDS);
// complete the callback-loop in case queue is cleared up
if (msg == null) {
if (log.isDebugEnabled()) {
@@ -883,10 +881,10 @@ public class ConsumerImpl extends ConsumerBase {
*
* @param message
*/
- void notifyPendingReceivedCallback(final MessageImpl message, Exception exception) {
+ void notifyPendingReceivedCallback(final Message<T> message, Exception exception) {
if (!pendingReceives.isEmpty()) {
// fetch receivedCallback from queue
- CompletableFuture<Message> receivedFuture = pendingReceives.poll();
+ CompletableFuture<Message<T>> receivedFuture = pendingReceives.poll();
if (exception == null) {
checkNotNull(message, "received message can't be null");
if (receivedFuture != null) {
@@ -950,8 +948,8 @@ public class ConsumerImpl extends ConsumerBase {
BatchMessageIdImpl batchMessageIdImpl = new BatchMessageIdImpl(messageId.getLedgerId(),
messageId.getEntryId(), getPartitionIndex(), i);
- final MessageImpl message = new MessageImpl(batchMessageIdImpl, msgMetadata,
- singleMessageMetadataBuilder.build(), singleMessagePayload, cnx);
+ final MessageImpl<T> message = new MessageImpl<>(batchMessageIdImpl, msgMetadata,
+ singleMessageMetadataBuilder.build(), singleMessagePayload, cnx, schema);
lock.readLock().lock();
try {
if (pendingReceives.isEmpty()) {
@@ -1384,7 +1382,7 @@ public class ConsumerImpl extends ConsumerBase {
private int removeExpiredMessagesFromQueue(Set<MessageId> messageIds) {
int messagesFromQueue = 0;
- Message peek = incomingMessages.peek();
+ Message<T> peek = incomingMessages.peek();
if (peek != null) {
MessageIdImpl messageId = getMessageIdImpl(peek);
if (!messageIds.contains(messageId)) {
@@ -1393,7 +1391,7 @@ public class ConsumerImpl extends ConsumerBase {
}
// try not to remove elements that are added while we remove
- Message message = incomingMessages.poll();
+ Message<T> message = incomingMessages.poll();
while (message != null) {
messagesFromQueue++;
MessageIdImpl id = getMessageIdImpl(message);
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageBuilderImpl.java
index c0dd67f..7714b99 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageBuilderImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageBuilderImpl.java
@@ -26,41 +26,52 @@ import java.util.Map;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageBuilder;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.api.proto.PulsarApi.KeyValue;
import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
import com.google.common.base.Preconditions;
-public class MessageBuilderImpl implements MessageBuilder {
+public class MessageBuilderImpl<T> implements MessageBuilder<T> {
private final MessageMetadata.Builder msgMetadataBuilder = MessageMetadata.newBuilder();
+ private final Schema<T> schema;
private ByteBuffer content;
+ public MessageBuilderImpl(Schema<T> schema) {
+ this.schema = schema;
+ }
+
+ @Override
+ public Message<T> build() {
+ return MessageImpl.create(msgMetadataBuilder, content, schema);
+ }
+
@Override
- public Message build() {
- return MessageImpl.create(msgMetadataBuilder, content);
+ public MessageBuilder<T> setValue(T value) {
+ return setContent(schema.encode(value));
}
@Override
- public MessageBuilder setContent(byte[] data) {
+ public MessageBuilder<T> setContent(byte[] data) {
setContent(data, 0, data.length);
return this;
}
@Override
- public MessageBuilder setContent(byte[] data, int offet, int length) {
+ public MessageBuilder<T> setContent(byte[] data, int offet, int length) {
this.content = ByteBuffer.wrap(data, offet, length);
return this;
}
@Override
- public MessageBuilder setContent(ByteBuffer buf) {
+ public MessageBuilder<T> setContent(ByteBuffer buf) {
this.content = buf.duplicate();
return this;
}
@Override
- public MessageBuilder setProperties(Map<String, String> properties) {
+ public MessageBuilder<T> setProperties(Map<String, String> properties) {
for (Map.Entry<String, String> entry : properties.entrySet()) {
msgMetadataBuilder
.addProperties(KeyValue.newBuilder().setKey(entry.getKey()).setValue(entry.getValue()).build());
@@ -70,33 +81,33 @@ public class MessageBuilderImpl implements MessageBuilder {
}
@Override
- public MessageBuilder setProperty(String name, String value) {
+ public MessageBuilder<T> setProperty(String name, String value) {
msgMetadataBuilder.addProperties(KeyValue.newBuilder().setKey(name).setValue(value).build());
return this;
}
@Override
- public MessageBuilder setKey(String key) {
+ public MessageBuilder<T> setKey(String key) {
msgMetadataBuilder.setPartitionKey(key);
return this;
}
@Override
- public MessageBuilder setEventTime(long timestamp) {
+ public MessageBuilder<T> setEventTime(long timestamp) {
checkArgument(timestamp > 0, "Invalid timestamp : '%s'", timestamp);
msgMetadataBuilder.setEventTime(timestamp);
return this;
}
@Override
- public MessageBuilder setSequenceId(long sequenceId) {
+ public MessageBuilder<T> setSequenceId(long sequenceId) {
checkArgument(sequenceId >= 0);
msgMetadataBuilder.setSequenceId(sequenceId);
return this;
}
@Override
- public MessageBuilder setReplicationClusters(List<String> clusters) {
+ public MessageBuilder<T> setReplicationClusters(List<String> clusters) {
Preconditions.checkNotNull(clusters);
msgMetadataBuilder.clearReplicateTo();
msgMetadataBuilder.addAllReplicateTo(clusters);
@@ -104,7 +115,7 @@ public class MessageBuilderImpl implements MessageBuilder {
}
@Override
- public MessageBuilder disableReplication() {
+ public MessageBuilder<T> disableReplication() {
msgMetadataBuilder.clearReplicateTo();
msgMetadataBuilder.addReplicateTo("__local__");
return this;
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java
index 5edf9e9..f4585ce 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessageImpl.java
@@ -30,6 +30,7 @@ import java.util.stream.Collectors;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.api.Commands;
import org.apache.pulsar.common.api.proto.PulsarApi;
import org.apache.pulsar.common.api.proto.PulsarApi.KeyValue;
@@ -43,29 +44,42 @@ import io.netty.buffer.Unpooled;
import io.netty.util.Recycler;
import io.netty.util.Recycler.Handle;
-public class MessageImpl implements Message {
+public class MessageImpl<T> implements Message<T> {
private MessageMetadata.Builder msgMetadataBuilder;
private MessageId messageId;
private ClientCnx cnx;
private ByteBuf payload;
+ private Schema<T> schema;
transient private Map<String, String> properties;
// Constructor for out-going message
- static MessageImpl create(MessageMetadata.Builder msgMetadataBuilder, ByteBuffer payload) {
- MessageImpl msg = RECYCLER.get();
+ static <T> MessageImpl<T> create(MessageMetadata.Builder msgMetadataBuilder, ByteBuffer payload, Schema<T> schema) {
+ MessageImpl<T> msg = RECYCLER.get();
+ msg.msgMetadataBuilder = msgMetadataBuilder;
+ msg.messageId = null;
+ msg.cnx = null;
+ msg.payload = Unpooled.wrappedBuffer(payload);
+ msg.properties = null;
+ msg.schema = schema;
+ return msg;
+ }
+
+ static MessageImpl<byte[]> create(MessageMetadata.Builder msgMetadataBuilder, ByteBuffer payload) {
+ MessageImpl<byte[]> msg = RECYCLER.get();
msg.msgMetadataBuilder = msgMetadataBuilder;
msg.messageId = null;
msg.cnx = null;
msg.payload = Unpooled.wrappedBuffer(payload);
msg.properties = null;
+ msg.schema = Schema.IDENTITY;
return msg;
}
// Constructor for incoming message
MessageImpl(MessageIdData messageId, MessageMetadata msgMetadata, ByteBuf payload, int partitionIndex,
- ClientCnx cnx) {
+ ClientCnx cnx, Schema<T> schema) {
this.msgMetadataBuilder = MessageMetadata.newBuilder(msgMetadata);
this.messageId = new MessageIdImpl(messageId.getLedgerId(), messageId.getEntryId(), partitionIndex);
this.cnx = cnx;
@@ -81,10 +95,11 @@ public class MessageImpl implements Message {
} else {
properties = Collections.emptyMap();
}
+ this.schema = schema;
}
MessageImpl(BatchMessageIdImpl batchMessageIdImpl, MessageMetadata msgMetadata,
- PulsarApi.SingleMessageMetadata singleMessageMetadata, ByteBuf payload, ClientCnx cnx) {
+ PulsarApi.SingleMessageMetadata singleMessageMetadata, ByteBuf payload, ClientCnx cnx, Schema<T> schema) {
this.msgMetadataBuilder = MessageMetadata.newBuilder(msgMetadata);
this.messageId = batchMessageIdImpl;
this.cnx = cnx;
@@ -104,13 +119,15 @@ public class MessageImpl implements Message {
if (singleMessageMetadata.hasPartitionKey()) {
msgMetadataBuilder.setPartitionKey(singleMessageMetadata.getPartitionKey());
}
+
+ this.schema = schema;
}
- public MessageImpl(String msgId, Map<String, String> properties, byte[] payload) {
- this(msgId, properties, Unpooled.wrappedBuffer(payload));
+ public MessageImpl(String msgId, Map<String, String> properties, byte[] payload, Schema<T> schema) {
+ this(msgId, properties, Unpooled.wrappedBuffer(payload), schema);
}
- public MessageImpl(String msgId, Map<String, String> properties, ByteBuf payload) {
+ public MessageImpl(String msgId, Map<String, String> properties, ByteBuf payload, Schema<T> schema) {
String[] data = msgId.split(":");
long ledgerId = Long.parseLong(data[0]);
long entryId = Long.parseLong(data[1]);
@@ -185,6 +202,10 @@ public class MessageImpl implements Message {
}
@Override
+ public T getValue() {
+ return schema.decode(getData());
+ }
+
public long getSequenceId() {
checkNotNull(msgMetadataBuilder);
if (msgMetadataBuilder.hasSequenceId()) {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedConsumerImpl.java
index 2fce234..f662ccf 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedConsumerImpl.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.client.impl;
import static com.google.common.base.Preconditions.checkArgument;
+import com.google.common.collect.Lists;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -33,11 +34,11 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.stream.Collectors;
-
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
@@ -46,11 +47,9 @@ import org.apache.pulsar.common.util.FutureUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.collect.Lists;
-
-public class PartitionedConsumerImpl extends ConsumerBase {
+public class PartitionedConsumerImpl<T> extends ConsumerBase<T> {
- private final List<ConsumerImpl> consumers;
+ private final List<ConsumerImpl<T>> consumers;
// Queue of partition consumers on which we have stopped calling receiveAsync() because the
// shared incoming queue was full
@@ -65,10 +64,10 @@ public class PartitionedConsumerImpl extends ConsumerBase {
private final ConsumerStats stats;
private final UnAckedMessageTracker unAckedMessageTracker;
- PartitionedConsumerImpl(PulsarClientImpl client, ConsumerConfigurationData conf, int numPartitions,
- ExecutorService listenerExecutor, CompletableFuture<Consumer> subscribeFuture) {
+ PartitionedConsumerImpl(PulsarClientImpl client, ConsumerConfigurationData<T> conf, int numPartitions,
+ ExecutorService listenerExecutor, CompletableFuture<Consumer<T>> subscribeFuture, Schema<T> schema) {
super(client, conf.getSingleTopic(), conf, Math.max(Math.max(2, numPartitions), conf.getReceiverQueueSize()),
- listenerExecutor, subscribeFuture);
+ listenerExecutor, subscribeFuture, schema);
this.consumers = Lists.newArrayListWithCapacity(numPartitions);
this.pausedConsumers = new ConcurrentLinkedQueue<>();
this.sharedQueueResumeThreshold = maxReceiverQueueSize / 2;
@@ -89,11 +88,11 @@ public class PartitionedConsumerImpl extends ConsumerBase {
private void start() {
AtomicReference<Throwable> subscribeFail = new AtomicReference<Throwable>();
AtomicInteger completed = new AtomicInteger();
- ConsumerConfigurationData internalConfig = getInternalConsumerConfig();
+ ConsumerConfigurationData<T> internalConfig = getInternalConsumerConfig();
for (int partitionIndex = 0; partitionIndex < numPartitions; partitionIndex++) {
String partitionName = TopicName.get(topic).getPartition(partitionIndex).toString();
- ConsumerImpl consumer = new ConsumerImpl(client, partitionName, internalConfig,
- client.externalExecutorProvider().getExecutor(), partitionIndex, new CompletableFuture<Consumer>());
+ ConsumerImpl<T> consumer = new ConsumerImpl<>(client, partitionName, internalConfig,
+ client.externalExecutorProvider().getExecutor(), partitionIndex, new CompletableFuture<>(), schema);
consumers.add(consumer);
consumer.subscribeFuture().handle((cons, subscribeException) -> {
if (subscribeException != null) {
@@ -128,13 +127,13 @@ public class PartitionedConsumerImpl extends ConsumerBase {
}
private void starReceivingMessages() throws PulsarClientException {
- for (ConsumerImpl consumer : consumers) {
+ for (ConsumerImpl<T> consumer : consumers) {
consumer.sendFlowPermitsToBroker(consumer.cnx(), conf.getReceiverQueueSize());
receiveMessageFromConsumer(consumer);
}
}
- private void receiveMessageFromConsumer(ConsumerImpl consumer) {
+ private void receiveMessageFromConsumer(ConsumerImpl<T> consumer) {
consumer.receiveAsync().thenAccept(message -> {
// Process the message, add to the queue and trigger listener or async callback
messageReceived(message);
@@ -183,8 +182,8 @@ public class PartitionedConsumerImpl extends ConsumerBase {
}
@Override
- protected Message internalReceive() throws PulsarClientException {
- Message message;
+ protected Message<T> internalReceive() throws PulsarClientException {
+ Message<T> message;
try {
message = incomingMessages.take();
unAckedMessageTracker.add((MessageIdImpl) message.getMessageId());
@@ -197,12 +196,12 @@ public class PartitionedConsumerImpl extends ConsumerBase {
}
@Override
- protected Message internalReceive(int timeout, TimeUnit unit) throws PulsarClientException {
- Message message;
+ protected Message<T> internalReceive(int timeout, TimeUnit unit) throws PulsarClientException {
+ Message<T> message;
try {
message = incomingMessages.poll(timeout, unit);
if (message != null) {
- unAckedMessageTracker.add((MessageIdImpl) message.getMessageId());
+ unAckedMessageTracker.add(message.getMessageId());
}
resumeReceivingFromPausedConsumersIfNeeded();
return message;
@@ -213,16 +212,16 @@ public class PartitionedConsumerImpl extends ConsumerBase {
}
@Override
- protected CompletableFuture<Message> internalReceiveAsync() {
- CompletableFuture<Message> result = new CompletableFuture<>();
- Message message;
+ protected CompletableFuture<Message<T>> internalReceiveAsync() {
+ CompletableFuture<Message<T>> result = new CompletableFuture<>();
+ Message<T> message;
try {
lock.writeLock().lock();
message = incomingMessages.poll(0, TimeUnit.SECONDS);
if (message == null) {
pendingReceives.add(result);
} else {
- unAckedMessageTracker.add((MessageIdImpl) message.getMessageId());
+ unAckedMessageTracker.add(message.getMessageId());
resumeReceivingFromPausedConsumersIfNeeded();
result.complete(message);
}
@@ -250,9 +249,9 @@ public class PartitionedConsumerImpl extends ConsumerBase {
"Cumulative acknowledge not supported for partitioned topics"));
} else {
- ConsumerImpl consumer = consumers.get(((MessageIdImpl) messageId).getPartitionIndex());
+ ConsumerImpl<T> consumer = consumers.get(((MessageIdImpl) messageId).getPartitionIndex());
return consumer.doAcknowledge(messageId, ackType, properties).thenRun(() ->
- unAckedMessageTracker.remove((MessageIdImpl) messageId));
+ unAckedMessageTracker.remove(messageId));
}
}
@@ -268,7 +267,7 @@ public class PartitionedConsumerImpl extends ConsumerBase {
AtomicReference<Throwable> unsubscribeFail = new AtomicReference<Throwable>();
AtomicInteger completed = new AtomicInteger(numPartitions);
CompletableFuture<Void> unsubscribeFuture = new CompletableFuture<>();
- for (Consumer consumer : consumers) {
+ for (Consumer<T> consumer : consumers) {
if (consumer != null) {
consumer.unsubscribeAsync().handle((unsubscribed, ex) -> {
if (ex != null) {
@@ -308,7 +307,7 @@ public class PartitionedConsumerImpl extends ConsumerBase {
AtomicReference<Throwable> closeFail = new AtomicReference<Throwable>();
AtomicInteger completed = new AtomicInteger(numPartitions);
CompletableFuture<Void> closeFuture = new CompletableFuture<>();
- for (Consumer consumer : consumers) {
+ for (Consumer<T> consumer : consumers) {
if (consumer != null) {
consumer.closeAsync().handle((closed, ex) -> {
if (ex != null) {
@@ -345,7 +344,7 @@ public class PartitionedConsumerImpl extends ConsumerBase {
try {
if (listenerExecutor != null && !listenerExecutor.isShutdown()) {
while (!pendingReceives.isEmpty()) {
- CompletableFuture<Message> receiveFuture = pendingReceives.poll();
+ CompletableFuture<Message<T>> receiveFuture = pendingReceives.poll();
if (receiveFuture != null) {
receiveFuture.completeExceptionally(
new PulsarClientException.AlreadyClosedException("Consumer is already closed"));
@@ -381,16 +380,16 @@ public class PartitionedConsumerImpl extends ConsumerBase {
}
- void messageReceived(Message message) {
+ void messageReceived(Message<T> message) {
lock.writeLock().lock();
try {
- unAckedMessageTracker.add((MessageIdImpl) message.getMessageId());
+ unAckedMessageTracker.add(message.getMessageId());
if (log.isDebugEnabled()) {
log.debug("[{}][{}] Received message from partitioned-consumer {}", topic, subscription, message.getMessageId());
}
// if asyncReceive is waiting : return message to callback without adding to incomingMessages queue
if (!pendingReceives.isEmpty()) {
- CompletableFuture<Message> receivedFuture = pendingReceives.poll();
+ CompletableFuture<Message<T>> receivedFuture = pendingReceives.poll();
listenerExecutor.execute(() -> receivedFuture.complete(message));
} else {
// Enqueue the message so that it can be retrieved when application calls receive()
@@ -408,7 +407,7 @@ public class PartitionedConsumerImpl extends ConsumerBase {
// Trigger the notification on the message listener in a separate thread to avoid blocking the networking
// thread while the message processing happens
listenerExecutor.execute(() -> {
- Message msg;
+ Message<T> msg;
try {
msg = internalReceive();
} catch (PulsarClientException e) {
@@ -434,8 +433,8 @@ public class PartitionedConsumerImpl extends ConsumerBase {
return subscription;
}
- private ConsumerConfigurationData getInternalConsumerConfig() {
- ConsumerConfigurationData internalConsumerConfig = new ConsumerConfigurationData();
+ private ConsumerConfigurationData<T> getInternalConsumerConfig() {
+ ConsumerConfigurationData<T> internalConsumerConfig = new ConsumerConfigurationData<>();
internalConsumerConfig.setReceiverQueueSize(conf.getReceiverQueueSize());
internalConsumerConfig.setSubscriptionName(conf.getSubscriptionName());
internalConsumerConfig.setSubscriptionType(conf.getSubscriptionType());
@@ -517,7 +516,7 @@ public class PartitionedConsumerImpl extends ConsumerBase {
return state;
}
- List<ConsumerImpl> getConsumers() {
+ List<ConsumerImpl<T>> getConsumers() {
return consumers;
}
@@ -553,7 +552,7 @@ public class PartitionedConsumerImpl extends ConsumerBase {
}
private void removeExpiredMessagesFromQueue(Set<MessageId> messageIds) {
- Message peek = incomingMessages.peek();
+ Message<T> peek = incomingMessages.peek();
if (peek != null) {
if (!messageIds.contains(peek.getMessageId())) {
// first message is not expired, then no message is expired in queue.
@@ -561,7 +560,7 @@ public class PartitionedConsumerImpl extends ConsumerBase {
}
// try not to remove elements that are added while we remove
- Message message = incomingMessages.poll();
+ Message<T> message = incomingMessages.poll();
while (message != null) {
MessageIdImpl messageId = (MessageIdImpl) message.getMessageId();
if (!messageIds.contains(messageId)) {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java
index 527df48..e70ce4b 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PartitionedProducerImpl.java
@@ -33,6 +33,7 @@ import org.apache.pulsar.client.api.MessageRouter;
import org.apache.pulsar.client.api.MessageRoutingMode;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.TopicMetadata;
import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
import org.apache.pulsar.common.naming.TopicName;
@@ -42,16 +43,16 @@ import org.slf4j.LoggerFactory;
import com.google.common.collect.Lists;
-public class PartitionedProducerImpl extends ProducerBase {
+public class PartitionedProducerImpl<T> extends ProducerBase<T> {
- private List<ProducerImpl> producers;
+ private List<ProducerImpl<T>> producers;
private MessageRouter routerPolicy;
private final ProducerStats stats;
private final TopicMetadata topicMetadata;
public PartitionedProducerImpl(PulsarClientImpl client, String topic, ProducerConfigurationData conf, int numPartitions,
- CompletableFuture<Producer> producerCreatedFuture) {
- super(client, topic, conf, producerCreatedFuture);
+ CompletableFuture<Producer<T>> producerCreatedFuture, Schema<T> schema) {
+ super(client, topic, conf, producerCreatedFuture, schema);
this.producers = Lists.newArrayListWithCapacity(numPartitions);
this.topicMetadata = new TopicMetadataImpl(numPartitions);
this.routerPolicy = getMessageRouter();
@@ -103,8 +104,8 @@ public class PartitionedProducerImpl extends ProducerBase {
AtomicInteger completed = new AtomicInteger();
for (int partitionIndex = 0; partitionIndex < topicMetadata.numPartitions(); partitionIndex++) {
String partitionName = TopicName.get(topic).getPartition(partitionIndex).toString();
- ProducerImpl producer = new ProducerImpl(client, partitionName, conf, new CompletableFuture<Producer>(),
- partitionIndex);
+ ProducerImpl<T> producer = new ProducerImpl<>(client, partitionName, conf, new CompletableFuture<>(),
+ partitionIndex, schema);
producers.add(producer);
producer.producerCreatedFuture().handle((prod, createException) -> {
if (createException != null) {
@@ -138,7 +139,7 @@ public class PartitionedProducerImpl extends ProducerBase {
}
@Override
- public CompletableFuture<MessageId> sendAsync(Message message) {
+ public CompletableFuture<MessageId> sendAsync(Message<T> message) {
switch (getState()) {
case Ready:
@@ -180,7 +181,7 @@ public class PartitionedProducerImpl extends ProducerBase {
AtomicReference<Throwable> closeFail = new AtomicReference<Throwable>();
AtomicInteger completed = new AtomicInteger(topicMetadata.numPartitions());
CompletableFuture<Void> closeFuture = new CompletableFuture<>();
- for (Producer producer : producers) {
+ for (Producer<T> producer : producers) {
if (producer != null) {
producer.closeAsync().handle((closed, ex) -> {
if (ex != null) {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternTopicsConsumerImpl.java
index 04445ea..f9cf550 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternTopicsConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PatternTopicsConsumerImpl.java
@@ -33,6 +33,7 @@ import java.util.concurrent.TimeUnit;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
@@ -40,17 +41,18 @@ import org.apache.pulsar.common.util.FutureUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class PatternTopicsConsumerImpl extends TopicsConsumerImpl implements TimerTask {
+public class PatternTopicsConsumerImpl<T> extends TopicsConsumerImpl<T> implements TimerTask {
private final Pattern topicsPattern;
private final TopicsChangedListener topicsChangeListener;
private volatile Timeout recheckPatternTimeout = null;
public PatternTopicsConsumerImpl(Pattern topicsPattern,
- PulsarClientImpl client,
- ConsumerConfigurationData conf,
- ExecutorService listenerExecutor,
- CompletableFuture<Consumer> subscribeFuture) {
- super(client, conf, listenerExecutor, subscribeFuture);
+ PulsarClientImpl client,
+ ConsumerConfigurationData<T> conf,
+ ExecutorService listenerExecutor,
+ CompletableFuture<Consumer<T>> subscribeFuture,
+ Schema<T> schema) {
+ super(client, conf, listenerExecutor, subscribeFuture, schema);
this.topicsPattern = topicsPattern;
if (this.namespaceName == null) {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBase.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBase.java
index 0876db1..14a9af0 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBase.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBase.java
@@ -27,33 +27,36 @@ import org.apache.pulsar.client.api.MessageBuilder;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
-public abstract class ProducerBase extends HandlerBase implements Producer {
+public abstract class ProducerBase<T> extends HandlerBase implements Producer<T> {
- protected final CompletableFuture<Producer> producerCreatedFuture;
+ protected final CompletableFuture<Producer<T>> producerCreatedFuture;
protected final ProducerConfigurationData conf;
+ protected final Schema<T> schema;
protected ProducerBase(PulsarClientImpl client, String topic, ProducerConfigurationData conf,
- CompletableFuture<Producer> producerCreatedFuture) {
+ CompletableFuture<Producer<T>> producerCreatedFuture, Schema<T> schema) {
super(client, topic, new Backoff(100, TimeUnit.MILLISECONDS, 60, TimeUnit.SECONDS,
Math.max(100, conf.getSendTimeoutMs() - 100), TimeUnit.MILLISECONDS));
this.producerCreatedFuture = producerCreatedFuture;
this.conf = conf;
+ this.schema = schema;
}
@Override
- public MessageId send(byte[] message) throws PulsarClientException {
- return send(MessageBuilder.create().setContent(message).build());
+ public MessageId send(T message) throws PulsarClientException {
+ return send(MessageBuilder.create(schema).setValue(message).build());
}
@Override
- public CompletableFuture<MessageId> sendAsync(byte[] message) {
- return sendAsync(MessageBuilder.create().setContent(message).build());
+ public CompletableFuture<MessageId> sendAsync(T message) {
+ return sendAsync(MessageBuilder.create(schema).setValue(message).build());
}
@Override
- public MessageId send(Message message) throws PulsarClientException {
+ public MessageId send(Message<T> message) throws PulsarClientException {
try {
return sendAsync(message).get();
} catch (ExecutionException e) {
@@ -70,7 +73,7 @@ public abstract class ProducerBase extends HandlerBase implements Producer {
}
@Override
- abstract public CompletableFuture<MessageId> sendAsync(Message message);
+ abstract public CompletableFuture<MessageId> sendAsync(Message<T> message);
@Override
public void close() throws PulsarClientException {
@@ -103,7 +106,7 @@ public abstract class ProducerBase extends HandlerBase implements Producer {
return conf;
}
- public CompletableFuture<Producer> producerCreatedFuture() {
+ public CompletableFuture<Producer<T>> producerCreatedFuture() {
return producerCreatedFuture;
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java
index 230d9c9..7ef878d 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerBuilderImpl.java
@@ -32,32 +32,35 @@ import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerBuilder;
import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
import org.apache.pulsar.common.util.FutureUtil;
-public class ProducerBuilderImpl implements ProducerBuilder {
+public class ProducerBuilderImpl<T> implements ProducerBuilder<T> {
private static final long serialVersionUID = 1L;
private final PulsarClientImpl client;
private final ProducerConfigurationData conf;
+ private final Schema<T> schema;
- ProducerBuilderImpl(PulsarClientImpl client) {
- this(client, new ProducerConfigurationData());
+ ProducerBuilderImpl(PulsarClientImpl client, Schema<T> schema) {
+ this(client, new ProducerConfigurationData(), schema);
}
- private ProducerBuilderImpl(PulsarClientImpl client, ProducerConfigurationData conf) {
+ private ProducerBuilderImpl(PulsarClientImpl client, ProducerConfigurationData conf, Schema<T> schema) {
this.client = client;
this.conf = conf;
+ this.schema = schema;
}
@Override
- public ProducerBuilder clone() {
- return new ProducerBuilderImpl(client, conf.clone());
+ public ProducerBuilder<T> clone() {
+ return new ProducerBuilderImpl<>(client, conf.clone(), schema);
}
@Override
- public Producer create() throws PulsarClientException {
+ public Producer<T> create() throws PulsarClientException {
try {
return createAsync().get();
} catch (ExecutionException e) {
@@ -74,125 +77,125 @@ public class ProducerBuilderImpl implements ProducerBuilder {
}
@Override
- public CompletableFuture<Producer> createAsync() {
+ public CompletableFuture<Producer<T>> createAsync() {
if (conf.getTopicName() == null) {
return FutureUtil
.failedFuture(new IllegalArgumentException("Topic name must be set on the producer builder"));
}
- return client.createProducerAsync(conf);
+ return client.createProducerAsync(conf, schema);
}
@Override
- public ProducerBuilder topic(String topicName) {
+ public ProducerBuilder<T> topic(String topicName) {
conf.setTopicName(topicName);
return this;
}
@Override
- public ProducerBuilder producerName(String producerName) {
+ public ProducerBuilder<T> producerName(String producerName) {
conf.setProducerName(producerName);
return this;
}
@Override
- public ProducerBuilder sendTimeout(int sendTimeout, TimeUnit unit) {
+ public ProducerBuilder<T> sendTimeout(int sendTimeout, TimeUnit unit) {
conf.setSendTimeoutMs(unit.toMillis(sendTimeout));
return this;
}
@Override
- public ProducerBuilder maxPendingMessages(int maxPendingMessages) {
+ public ProducerBuilder<T> maxPendingMessages(int maxPendingMessages) {
conf.setMaxPendingMessages(maxPendingMessages);
return this;
}
@Override
- public ProducerBuilder maxPendingMessagesAcrossPartitions(int maxPendingMessagesAcrossPartitions) {
+ public ProducerBuilder<T> maxPendingMessagesAcrossPartitions(int maxPendingMessagesAcrossPartitions) {
conf.setMaxPendingMessagesAcrossPartitions(maxPendingMessagesAcrossPartitions);
return this;
}
@Override
- public ProducerBuilder blockIfQueueFull(boolean blockIfQueueFull) {
+ public ProducerBuilder<T> blockIfQueueFull(boolean blockIfQueueFull) {
conf.setBlockIfQueueFull(blockIfQueueFull);
return this;
}
@Override
- public ProducerBuilder messageRoutingMode(MessageRoutingMode messageRouteMode) {
+ public ProducerBuilder<T> messageRoutingMode(MessageRoutingMode messageRouteMode) {
conf.setMessageRoutingMode(messageRouteMode);
return this;
}
@Override
- public ProducerBuilder compressionType(CompressionType compressionType) {
+ public ProducerBuilder<T> compressionType(CompressionType compressionType) {
conf.setCompressionType(compressionType);
return this;
}
@Override
- public ProducerBuilder hashingScheme(HashingScheme hashingScheme) {
+ public ProducerBuilder<T> hashingScheme(HashingScheme hashingScheme) {
conf.setHashingScheme(hashingScheme);
return this;
}
@Override
- public ProducerBuilder messageRouter(MessageRouter messageRouter) {
+ public ProducerBuilder<T> messageRouter(MessageRouter messageRouter) {
conf.setCustomMessageRouter(messageRouter);
return this;
}
@Override
- public ProducerBuilder enableBatching(boolean batchMessagesEnabled) {
+ public ProducerBuilder<T> enableBatching(boolean batchMessagesEnabled) {
conf.setBatchingEnabled(batchMessagesEnabled);
return this;
}
@Override
- public ProducerBuilder cryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
+ public ProducerBuilder<T> cryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
conf.setCryptoKeyReader(cryptoKeyReader);
return this;
}
@Override
- public ProducerBuilder addEncryptionKey(String key) {
+ public ProducerBuilder<T> addEncryptionKey(String key) {
conf.getEncryptionKeys().add(key);
return this;
}
@Override
- public ProducerBuilder cryptoFailureAction(ProducerCryptoFailureAction action) {
+ public ProducerBuilder<T> cryptoFailureAction(ProducerCryptoFailureAction action) {
conf.setCryptoFailureAction(action);
return this;
}
@Override
- public ProducerBuilder batchingMaxPublishDelay(long batchDelay, TimeUnit timeUnit) {
+ public ProducerBuilder<T> batchingMaxPublishDelay(long batchDelay, TimeUnit timeUnit) {
conf.setBatchingMaxPublishDelayMicros(timeUnit.toMicros(batchDelay));
return this;
}
@Override
- public ProducerBuilder batchingMaxMessages(int batchMessagesMaxMessagesPerBatch) {
+ public ProducerBuilder<T> batchingMaxMessages(int batchMessagesMaxMessagesPerBatch) {
conf.setBatchingMaxMessages(batchMessagesMaxMessagesPerBatch);
return this;
}
@Override
- public ProducerBuilder initialSequenceId(long initialSequenceId) {
+ public ProducerBuilder<T> initialSequenceId(long initialSequenceId) {
conf.setInitialSequenceId(initialSequenceId);
return this;
}
@Override
- public ProducerBuilder property(String key, String value) {
+ public ProducerBuilder<T> property(String key, String value) {
conf.getProperties().put(key, value);
return this;
}
@Override
- public ProducerBuilder properties(Map<String, String> properties) {
+ public ProducerBuilder<T> properties(Map<String, String> properties) {
conf.getProperties().putAll(properties);
return this;
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
index c94baae..b55f279 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java
@@ -44,6 +44,7 @@ import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.PulsarClientException.CryptoException;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.conf.ProducerConfigurationData;
import org.apache.pulsar.common.api.ByteBufPair;
import org.apache.pulsar.common.api.Commands;
@@ -68,7 +69,7 @@ import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import io.netty.util.concurrent.ScheduledFuture;
-public class ProducerImpl extends ProducerBase implements TimerTask {
+public class ProducerImpl<T> extends ProducerBase<T> implements TimerTask {
// Producer id, used to identify a producer within a single connection
private final long producerId;
@@ -107,8 +108,8 @@ public class ProducerImpl extends ProducerBase implements TimerTask {
.newUpdater(ProducerImpl.class, "msgIdGenerator");
public ProducerImpl(PulsarClientImpl client, String topic, ProducerConfigurationData conf,
- CompletableFuture<Producer> producerCreatedFuture, int partitionIndex) {
- super(client, topic, conf, producerCreatedFuture);
+ CompletableFuture<Producer<T>> producerCreatedFuture, int partitionIndex, Schema<T> schema) {
+ super(client, topic, conf, producerCreatedFuture, schema);
this.producerId = client.newProducerId();
this.producerName = conf.getProducerName();
this.partitionIndex = partitionIndex;
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
index fd3cada..1741eeb 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java
@@ -18,9 +18,15 @@
*/
package org.apache.pulsar.client.impl;
-import static org.apache.commons.lang3.StringUtils.isBlank;
import static com.google.common.base.Preconditions.checkState;
+import static org.apache.commons.lang3.StringUtils.isBlank;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import io.netty.channel.EventLoopGroup;
+import io.netty.util.HashedWheelTimer;
+import io.netty.util.Timer;
+import io.netty.util.concurrent.DefaultThreadFactory;
import java.util.IdentityHashMap;
import java.util.List;
import java.util.concurrent.CompletableFuture;
@@ -46,6 +52,7 @@ import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Reader;
import org.apache.pulsar.client.api.ReaderBuilder;
import org.apache.pulsar.client.api.ReaderConfiguration;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
@@ -61,14 +68,6 @@ import org.apache.pulsar.common.util.netty.EventLoopUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-
-import io.netty.channel.EventLoopGroup;
-import io.netty.util.HashedWheelTimer;
-import io.netty.util.Timer;
-import io.netty.util.concurrent.DefaultThreadFactory;
-
@SuppressWarnings("deprecation")
public class PulsarClientImpl implements PulsarClient {
@@ -85,8 +84,8 @@ public class PulsarClientImpl implements PulsarClient {
}
private AtomicReference<State> state = new AtomicReference<>();
- private final IdentityHashMap<ProducerBase, Boolean> producers;
- private final IdentityHashMap<ConsumerBase, Boolean> consumers;
+ private final IdentityHashMap<ProducerBase<?>, Boolean> producers;
+ private final IdentityHashMap<ConsumerBase<?>, Boolean> consumers;
private final AtomicLong producerIdGenerator = new AtomicLong();
private final AtomicLong consumerIdGenerator = new AtomicLong();
@@ -145,22 +144,37 @@ public class PulsarClientImpl implements PulsarClient {
}
@Override
- public ProducerBuilder newProducer() {
- return new ProducerBuilderImpl(this);
+ public ProducerBuilder<byte[]> newProducer() {
+ return new ProducerBuilderImpl<>(this, Schema.IDENTITY);
}
@Override
- public ConsumerBuilder newConsumer() {
- return new ConsumerBuilderImpl(this);
+ public <T> ProducerBuilder<T> newProducer(Schema<T> schema) {
+ return new ProducerBuilderImpl<>(this, schema);
}
@Override
- public ReaderBuilder newReader() {
- return new ReaderBuilderImpl(this);
+ public ConsumerBuilder<byte[]> newConsumer() {
+ return new ConsumerBuilderImpl<>(this, Schema.IDENTITY);
}
@Override
- public Producer createProducer(String topic) throws PulsarClientException {
+ public <T> ConsumerBuilder<T> newConsumer(Schema<T> schema) {
+ return new ConsumerBuilderImpl<>(this, schema);
+ }
+
+ @Override
+ public ReaderBuilder<byte[]> newReader() {
+ return new ReaderBuilderImpl<>(this, Schema.IDENTITY);
+ }
+
+ @Override
+ public <T> ReaderBuilder<T> newReader(Schema<T> schema) {
+ return new ReaderBuilderImpl<>(this, schema);
+ }
+
+ @Override
+ public Producer<byte[]> createProducer(String topic) throws PulsarClientException {
try {
ProducerConfigurationData conf = new ProducerConfigurationData();
conf.setTopicName(topic);
@@ -179,7 +193,7 @@ public class PulsarClientImpl implements PulsarClient {
}
@Override
- public Producer createProducer(final String topic, final ProducerConfiguration conf) throws PulsarClientException {
+ public Producer<byte[]> createProducer(final String topic, final ProducerConfiguration conf) throws PulsarClientException {
if (conf == null) {
throw new PulsarClientException.InvalidConfigurationException("Invalid null configuration object");
}
@@ -202,20 +216,24 @@ public class PulsarClientImpl implements PulsarClient {
}
@Override
- public CompletableFuture<Producer> createProducerAsync(String topic) {
+ public CompletableFuture<Producer<byte[]>> createProducerAsync(String topic) {
ProducerConfigurationData conf = new ProducerConfigurationData();
conf.setTopicName(topic);
return createProducerAsync(conf);
}
@Override
- public CompletableFuture<Producer> createProducerAsync(final String topic, final ProducerConfiguration conf) {
+ public CompletableFuture<Producer<byte[]>> createProducerAsync(final String topic, final ProducerConfiguration conf) {
ProducerConfigurationData confData = conf.getProducerConfigurationData().clone();
confData.setTopicName(topic);
return createProducerAsync(confData);
}
- public CompletableFuture<Producer> createProducerAsync(ProducerConfigurationData conf) {
+ public CompletableFuture<Producer<byte[]>> createProducerAsync(ProducerConfigurationData conf) {
+ return createProducerAsync(conf, Schema.IDENTITY);
+ }
+
+ public <T> CompletableFuture<Producer<T>> createProducerAsync(ProducerConfigurationData conf, Schema<T> schema) {
if (conf == null) {
return FutureUtil.failedFuture(
new PulsarClientException.InvalidConfigurationException("Producer configuration undefined"));
@@ -231,7 +249,7 @@ public class PulsarClientImpl implements PulsarClient {
return FutureUtil.failedFuture(new PulsarClientException.InvalidTopicNameException("Invalid topic name"));
}
- CompletableFuture<Producer> producerCreatedFuture = new CompletableFuture<>();
+ CompletableFuture<Producer<T>> producerCreatedFuture = new CompletableFuture<>();
getPartitionedTopicMetadata(topic).thenAccept(metadata -> {
if (log.isDebugEnabled()) {
@@ -240,10 +258,10 @@ public class PulsarClientImpl implements PulsarClient {
ProducerBase producer;
if (metadata.partitions > 1) {
- producer = new PartitionedProducerImpl(PulsarClientImpl.this, topic, conf, metadata.partitions,
- producerCreatedFuture);
+ producer = new PartitionedProducerImpl<>(PulsarClientImpl.this, topic, conf, metadata.partitions,
+ producerCreatedFuture, schema);
} else {
- producer = new ProducerImpl(PulsarClientImpl.this, topic, conf, producerCreatedFuture, -1);
+ producer = new ProducerImpl<>(PulsarClientImpl.this, topic, conf, producerCreatedFuture, -1, schema);
}
synchronized (producers) {
@@ -259,12 +277,12 @@ public class PulsarClientImpl implements PulsarClient {
}
@Override
- public Consumer subscribe(final String topic, final String subscription) throws PulsarClientException {
+ public Consumer<byte[]> subscribe(final String topic, final String subscription) throws PulsarClientException {
return subscribe(topic, subscription, new ConsumerConfiguration());
}
@Override
- public Consumer subscribe(String topic, String subscription, ConsumerConfiguration conf)
+ public Consumer<byte[]> subscribe(String topic, String subscription, ConsumerConfiguration conf)
throws PulsarClientException {
try {
return subscribeAsync(topic, subscription, conf).get();
@@ -282,28 +300,32 @@ public class PulsarClientImpl implements PulsarClient {
}
@Override
- public CompletableFuture<Consumer> subscribeAsync(String topic, String subscription) {
- ConsumerConfigurationData conf = new ConsumerConfigurationData();
+ public CompletableFuture<Consumer<byte[]>> subscribeAsync(String topic, String subscription) {
+ ConsumerConfigurationData<byte[]> conf = new ConsumerConfigurationData<>();
conf.getTopicNames().add(topic);
conf.setSubscriptionName(subscription);
return subscribeAsync(conf);
}
@Override
- public CompletableFuture<Consumer> subscribeAsync(final String topic, final String subscription,
+ public CompletableFuture<Consumer<byte[]>> subscribeAsync(final String topic, final String subscription,
final ConsumerConfiguration conf) {
if (conf == null) {
return FutureUtil.failedFuture(
new PulsarClientException.InvalidConfigurationException("Invalid null configuration"));
}
- ConsumerConfigurationData confData = conf.getConfigurationData().clone();
+ ConsumerConfigurationData<byte[]> confData = conf.getConfigurationData().clone();
confData.getTopicNames().add(topic);
confData.setSubscriptionName(subscription);
return subscribeAsync(confData);
}
- public CompletableFuture<Consumer> subscribeAsync(ConsumerConfigurationData conf) {
+ public CompletableFuture<Consumer<byte[]>> subscribeAsync(ConsumerConfigurationData<byte[]> conf) {
+ return subscribeAsync(conf, Schema.IDENTITY);
+ }
+
+ public <T> CompletableFuture<Consumer<T>> subscribeAsync(ConsumerConfigurationData<T> conf, Schema<T> schema) {
if (state.get() != State.Open) {
return FutureUtil.failedFuture(new PulsarClientException.AlreadyClosedException("Client already closed"));
}
@@ -313,7 +335,7 @@ public class PulsarClientImpl implements PulsarClient {
new PulsarClientException.InvalidConfigurationException("Consumer configuration undefined"));
}
- if (!conf.getTopicNames().stream().allMatch(topic -> TopicName.isValid(topic))) {
+ if (!conf.getTopicNames().stream().allMatch(TopicName::isValid)) {
return FutureUtil.failedFuture(new PulsarClientException.InvalidTopicNameException("Invalid topic name"));
}
@@ -341,16 +363,16 @@ public class PulsarClientImpl implements PulsarClient {
return FutureUtil
.failedFuture(new IllegalArgumentException("Topic names list must be null when use topicsPattern"));
}
- return patternTopicSubscribeAsync(conf);
+ return patternTopicSubscribeAsync(conf, schema);
} else if (conf.getTopicNames().size() == 1) {
- return singleTopicSubscribeAsysnc(conf);
+ return singleTopicSubscribeAsync(conf, schema);
} else {
- return multiTopicSubscribeAsync(conf);
+ return multiTopicSubscribeAsync(conf, schema);
}
}
- private CompletableFuture<Consumer> singleTopicSubscribeAsysnc(ConsumerConfigurationData conf) {
- CompletableFuture<Consumer> consumerSubscribedFuture = new CompletableFuture<>();
+ private <T> CompletableFuture<Consumer<T>> singleTopicSubscribeAsync(ConsumerConfigurationData<T> conf, Schema<T> schema) {
+ CompletableFuture<Consumer<T>> consumerSubscribedFuture = new CompletableFuture<>();
String topic = conf.getSingleTopic();
@@ -359,15 +381,15 @@ public class PulsarClientImpl implements PulsarClient {
log.debug("[{}] Received topic metadata. partitions: {}", topic, metadata.partitions);
}
- ConsumerBase consumer;
+ ConsumerBase<T> consumer;
// gets the next single threaded executor from the list of executors
ExecutorService listenerThread = externalExecutorProvider.getExecutor();
if (metadata.partitions > 1) {
- consumer = new PartitionedConsumerImpl(PulsarClientImpl.this, conf, metadata.partitions, listenerThread,
- consumerSubscribedFuture);
+ consumer = new PartitionedConsumerImpl<>(PulsarClientImpl.this, conf, metadata.partitions, listenerThread,
+ consumerSubscribedFuture, schema);
} else {
- consumer = new ConsumerImpl(PulsarClientImpl.this, topic, conf, listenerThread, -1,
- consumerSubscribedFuture);
+ consumer = new ConsumerImpl<>(PulsarClientImpl.this, topic, conf, listenerThread, -1,
+ consumerSubscribedFuture, schema);
}
synchronized (consumers) {
@@ -382,11 +404,11 @@ public class PulsarClientImpl implements PulsarClient {
return consumerSubscribedFuture;
}
- private CompletableFuture<Consumer> multiTopicSubscribeAsync(ConsumerConfigurationData conf) {
- CompletableFuture<Consumer> consumerSubscribedFuture = new CompletableFuture<>();
+ private <T> CompletableFuture<Consumer<T>> multiTopicSubscribeAsync(ConsumerConfigurationData<T> conf, Schema<T> schema) {
+ CompletableFuture<Consumer<T>> consumerSubscribedFuture = new CompletableFuture<>();
- ConsumerBase consumer = new TopicsConsumerImpl(PulsarClientImpl.this, conf,
- externalExecutorProvider.getExecutor(), consumerSubscribedFuture);
+ ConsumerBase<T> consumer = new TopicsConsumerImpl<>(PulsarClientImpl.this, conf,
+ externalExecutorProvider.getExecutor(), consumerSubscribedFuture, schema);
synchronized (consumers) {
consumers.put(consumer, Boolean.TRUE);
@@ -395,12 +417,16 @@ public class PulsarClientImpl implements PulsarClient {
return consumerSubscribedFuture;
}
- public CompletableFuture<Consumer> patternTopicSubscribeAsync(ConsumerConfigurationData conf) {
+ public CompletableFuture<Consumer<byte[]>> patternTopicSubscribeAsync(ConsumerConfigurationData<byte[]> conf) {
+ return patternTopicSubscribeAsync(conf, Schema.IDENTITY);
+ }
+
+ private <T> CompletableFuture<Consumer<T>> patternTopicSubscribeAsync(ConsumerConfigurationData<T> conf, Schema<T> schema) {
String regex = conf.getTopicsPattern().pattern();
TopicName destination = TopicName.get(regex);
NamespaceName namespaceName = destination.getNamespaceObject();
- CompletableFuture<Consumer> consumerSubscribedFuture = new CompletableFuture<>();
+ CompletableFuture<Consumer<T>> consumerSubscribedFuture = new CompletableFuture<>();
lookup.getTopicsUnderNamespace(namespaceName)
.thenAccept(topics -> {
if (log.isDebugEnabled()) {
@@ -411,11 +437,12 @@ public class PulsarClientImpl implements PulsarClient {
List<String> topicsList = topicsPatternFilter(topics, conf.getTopicsPattern());
conf.getTopicNames().addAll(topicsList);
- ConsumerBase consumer = new PatternTopicsConsumerImpl(conf.getTopicsPattern(),
+ ConsumerBase consumer = new PatternTopicsConsumerImpl<>(conf.getTopicsPattern(),
PulsarClientImpl.this,
conf,
externalExecutorProvider.getExecutor(),
- consumerSubscribedFuture);
+ consumerSubscribedFuture,
+ schema);
synchronized (consumers) {
consumers.put(consumer, Boolean.TRUE);
@@ -442,7 +469,7 @@ public class PulsarClientImpl implements PulsarClient {
}
@Override
- public Reader createReader(String topic, MessageId startMessageId, ReaderConfiguration conf)
+ public Reader<byte[]> createReader(String topic, MessageId startMessageId, ReaderConfiguration conf)
throws PulsarClientException {
try {
return createReaderAsync(topic, startMessageId, conf).get();
@@ -460,15 +487,19 @@ public class PulsarClientImpl implements PulsarClient {
}
@Override
- public CompletableFuture<Reader> createReaderAsync(String topic, MessageId startMessageId,
+ public CompletableFuture<Reader<byte[]>> createReaderAsync(String topic, MessageId startMessageId,
ReaderConfiguration conf) {
- ReaderConfigurationData confData = conf.getReaderConfigurationData().clone();
+ ReaderConfigurationData<byte[]> confData = conf.getReaderConfigurationData().clone();
confData.setTopicName(topic);
confData.setStartMessageId(startMessageId);
return createReaderAsync(confData);
}
- public CompletableFuture<Reader> createReaderAsync(ReaderConfigurationData conf) {
+ public CompletableFuture<Reader<byte[]>> createReaderAsync(ReaderConfigurationData<byte[]> conf) {
+ return createReaderAsync(conf, Schema.IDENTITY);
+ }
+
+ public <T> CompletableFuture<Reader<T>> createReaderAsync(ReaderConfigurationData<T> conf, Schema<T> schema) {
if (state.get() != State.Open) {
return FutureUtil.failedFuture(new PulsarClientException.AlreadyClosedException("Client already closed"));
}
@@ -489,7 +520,7 @@ public class PulsarClientImpl implements PulsarClient {
.failedFuture(new PulsarClientException.InvalidConfigurationException("Invalid startMessageId"));
}
- CompletableFuture<Reader> readerFuture = new CompletableFuture<>();
+ CompletableFuture<Reader<T>> readerFuture = new CompletableFuture<>();
getPartitionedTopicMetadata(topic).thenAccept(metadata -> {
if (log.isDebugEnabled()) {
@@ -502,10 +533,10 @@ public class PulsarClientImpl implements PulsarClient {
return;
}
- CompletableFuture<Consumer> consumerSubscribedFuture = new CompletableFuture<>();
+ CompletableFuture<Consumer<T>> consumerSubscribedFuture = new CompletableFuture<>();
// gets the next single threaded executor from the list of executors
ExecutorService listenerThread = externalExecutorProvider.getExecutor();
- ReaderImpl reader = new ReaderImpl(PulsarClientImpl.this, conf, listenerThread, consumerSubscribedFuture);
+ ReaderImpl<T> reader = new ReaderImpl<>(PulsarClientImpl.this, conf, listenerThread, consumerSubscribedFuture, schema);
synchronized (consumers) {
consumers.put(reader.getConsumer(), Boolean.TRUE);
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderBuilderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderBuilderImpl.java
index 9a537b1..bef909d 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderBuilderImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderBuilderImpl.java
@@ -28,37 +28,41 @@ import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Reader;
import org.apache.pulsar.client.api.ReaderBuilder;
import org.apache.pulsar.client.api.ReaderListener;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.impl.conf.ReaderConfigurationData;
import org.apache.pulsar.common.util.FutureUtil;
-public class ReaderBuilderImpl implements ReaderBuilder {
+public class ReaderBuilderImpl<T> implements ReaderBuilder<T> {
private static final long serialVersionUID = 1L;
private final PulsarClientImpl client;
- private final ReaderConfigurationData conf;
+ private final ReaderConfigurationData<T> conf;
- ReaderBuilderImpl(PulsarClientImpl client) {
- this(client, new ReaderConfigurationData());
+ private final Schema<T> schema;
+
+ ReaderBuilderImpl(PulsarClientImpl client, Schema<T> schema) {
+ this(client, new ReaderConfigurationData(), schema);
}
- private ReaderBuilderImpl(PulsarClientImpl client, ReaderConfigurationData conf) {
+ private ReaderBuilderImpl(PulsarClientImpl client, ReaderConfigurationData conf, Schema<T> schema) {
this.client = client;
this.conf = conf;
+ this.schema = schema;
}
@Override
- public ReaderBuilder clone() {
+ public ReaderBuilder<T> clone() {
try {
- return (ReaderBuilder) super.clone();
+ return (ReaderBuilder<T>) super.clone();
} catch (CloneNotSupportedException e) {
throw new RuntimeException("Failed to clone ReaderBuilderImpl");
}
}
@Override
- public Reader create() throws PulsarClientException {
+ public Reader<T> create() throws PulsarClientException {
try {
return createAsync().get();
} catch (ExecutionException e) {
@@ -75,7 +79,7 @@ public class ReaderBuilderImpl implements ReaderBuilder {
}
@Override
- public CompletableFuture<Reader> createAsync() {
+ public CompletableFuture<Reader<T>> createAsync() {
if (conf.getTopicName() == null) {
return FutureUtil
.failedFuture(new IllegalArgumentException("Topic name must be set on the reader builder"));
@@ -86,53 +90,53 @@ public class ReaderBuilderImpl implements ReaderBuilder {
.failedFuture(new IllegalArgumentException("Start message id must be set on the reader builder"));
}
- return client.createReaderAsync(conf);
+ return client.createReaderAsync(conf, schema);
}
@Override
- public ReaderBuilder topic(String topicName) {
+ public ReaderBuilder<T> topic(String topicName) {
conf.setTopicName(topicName);
return this;
}
@Override
- public ReaderBuilder startMessageId(MessageId startMessageId) {
+ public ReaderBuilder<T> startMessageId(MessageId startMessageId) {
conf.setStartMessageId(startMessageId);
return this;
}
@Override
- public ReaderBuilder readerListener(ReaderListener readerListener) {
+ public ReaderBuilder<T> readerListener(ReaderListener readerListener) {
conf.setReaderListener(readerListener);
return this;
}
@Override
- public ReaderBuilder cryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
+ public ReaderBuilder<T> cryptoKeyReader(CryptoKeyReader cryptoKeyReader) {
conf.setCryptoKeyReader(cryptoKeyReader);
return this;
}
@Override
- public ReaderBuilder cryptoFailureAction(ConsumerCryptoFailureAction action) {
+ public ReaderBuilder<T> cryptoFailureAction(ConsumerCryptoFailureAction action) {
conf.setCryptoFailureAction(action);
return this;
}
@Override
- public ReaderBuilder receiverQueueSize(int receiverQueueSize) {
+ public ReaderBuilder<T> receiverQueueSize(int receiverQueueSize) {
conf.setReceiverQueueSize(receiverQueueSize);
return this;
}
@Override
- public ReaderBuilder readerName(String readerName) {
+ public ReaderBuilder<T> readerName(String readerName) {
conf.setReaderName(readerName);
return this;
}
@Override
- public ReaderBuilder subscriptionRolePrefix(String subscriptionRolePrefix) {
+ public ReaderBuilder<T> subscriptionRolePrefix(String subscriptionRolePrefix) {
conf.setSubscriptionRolePrefix(subscriptionRolePrefix);
return this;
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java
index aaa1e4f..daf4799 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ReaderImpl.java
@@ -23,7 +23,6 @@ import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
-
import org.apache.commons.codec.digest.DigestUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.client.api.Consumer;
@@ -32,24 +31,25 @@ import org.apache.pulsar.client.api.MessageListener;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Reader;
import org.apache.pulsar.client.api.ReaderListener;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.ConsumerImpl.SubscriptionMode;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.apache.pulsar.client.impl.conf.ReaderConfigurationData;
-public class ReaderImpl implements Reader {
+public class ReaderImpl<T> implements Reader<T> {
- private final ConsumerImpl consumer;
+ private final ConsumerImpl<T> consumer;
- public ReaderImpl(PulsarClientImpl client, ReaderConfigurationData readerConfiguration,
- ExecutorService listenerExecutor, CompletableFuture<Consumer> consumerFuture) {
+ public ReaderImpl(PulsarClientImpl client, ReaderConfigurationData<T> readerConfiguration,
+ ExecutorService listenerExecutor, CompletableFuture<Consumer<T>> consumerFuture, Schema<T> schema) {
String subscription = "reader-" + DigestUtils.sha1Hex(UUID.randomUUID().toString()).substring(0, 10);
if (StringUtils.isNotBlank(readerConfiguration.getSubscriptionRolePrefix())) {
subscription = readerConfiguration.getSubscriptionRolePrefix() + "-" + subscription;
}
- ConsumerConfigurationData consumerConfiguration = new ConsumerConfigurationData();
+ ConsumerConfigurationData<T> consumerConfiguration = new ConsumerConfigurationData<>();
consumerConfiguration.getTopicNames().add(readerConfiguration.getTopicName());
consumerConfiguration.setSubscriptionName(subscription);
consumerConfiguration.setSubscriptionType(SubscriptionType.Exclusive);
@@ -59,18 +59,18 @@ public class ReaderImpl implements Reader {
}
if (readerConfiguration.getReaderListener() != null) {
- ReaderListener readerListener = readerConfiguration.getReaderListener();
- consumerConfiguration.setMessageListener(new MessageListener() {
+ ReaderListener<T> readerListener = readerConfiguration.getReaderListener();
+ consumerConfiguration.setMessageListener(new MessageListener<T>() {
private static final long serialVersionUID = 1L;
@Override
- public void received(Consumer consumer, Message msg) {
+ public void received(Consumer<T> consumer, Message<T> msg) {
readerListener.received(ReaderImpl.this, msg);
consumer.acknowledgeCumulativeAsync(msg);
}
@Override
- public void reachedEndOfTopic(Consumer consumer) {
+ public void reachedEndOfTopic(Consumer<T> consumer) {
readerListener.reachedEndOfTopic(ReaderImpl.this);
}
});
@@ -81,8 +81,8 @@ public class ReaderImpl implements Reader {
consumerConfiguration.setCryptoKeyReader(readerConfiguration.getCryptoKeyReader());
}
- consumer = new ConsumerImpl(client, readerConfiguration.getTopicName(), consumerConfiguration, listenerExecutor,
- -1, consumerFuture, SubscriptionMode.NonDurable, readerConfiguration.getStartMessageId());
+ consumer = new ConsumerImpl<>(client, readerConfiguration.getTopicName(), consumerConfiguration, listenerExecutor,
+ -1, consumerFuture, SubscriptionMode.NonDurable, readerConfiguration.getStartMessageId(), schema);
}
@Override
@@ -100,8 +100,8 @@ public class ReaderImpl implements Reader {
}
@Override
- public Message readNext() throws PulsarClientException {
- Message msg = consumer.receive();
+ public Message<T> readNext() throws PulsarClientException {
+ Message<T> msg = consumer.receive();
// Acknowledge message immediately because the reader is based on non-durable subscription. When it reconnects,
// it will specify the subscription position anyway
@@ -110,8 +110,8 @@ public class ReaderImpl implements Reader {
}
@Override
- public Message readNext(int timeout, TimeUnit unit) throws PulsarClientException {
- Message msg = consumer.receive(timeout, unit);
+ public Message<T> readNext(int timeout, TimeUnit unit) throws PulsarClientException {
+ Message<T> msg = consumer.receive(timeout, unit);
if (msg != null) {
consumer.acknowledgeCumulativeAsync(msg);
@@ -120,7 +120,7 @@ public class ReaderImpl implements Reader {
}
@Override
- public CompletableFuture<Message> readNextAsync() {
+ public CompletableFuture<Message<T>> readNextAsync() {
return consumer.receiveAsync().thenApply(msg -> {
consumer.acknowledgeCumulativeAsync(msg);
return msg;
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicMessageImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicMessageImpl.java
index 619d15c..86e5cf6 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicMessageImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicMessageImpl.java
@@ -98,4 +98,9 @@ public class TopicMessageImpl implements Message {
public String getKey() {
return msg.getKey();
}
+
+ @Override
+ public Object getValue() {
+ return msg.getData();
+ }
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicsConsumerImpl.java
index 4a8e663..a2c7b53 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicsConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TopicsConsumerImpl.java
@@ -42,6 +42,7 @@ import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.conf.ConsumerConfigurationData;
import org.apache.pulsar.client.util.ConsumerName;
@@ -54,20 +55,20 @@ import org.slf4j.LoggerFactory;
import com.google.common.collect.Lists;
-public class TopicsConsumerImpl extends ConsumerBase {
+public class TopicsConsumerImpl<T> extends ConsumerBase<T> {
// All topics should be in same namespace
protected NamespaceName namespaceName;
// Map <topic+partition, consumer>, when get do ACK, consumer will by find by topic name
- private final ConcurrentHashMap<String, ConsumerImpl> consumers;
+ private final ConcurrentHashMap<String, ConsumerImpl<T>> consumers;
// Map <topic, partitionNumber>, store partition number for each topic
protected final ConcurrentHashMap<String, Integer> topics;
// Queue of partition consumers on which we have stopped calling receiveAsync() because the
// shared incoming queue was full
- private final ConcurrentLinkedQueue<ConsumerImpl> pausedConsumers;
+ private final ConcurrentLinkedQueue<ConsumerImpl<T>> pausedConsumers;
// Threshold for the shared queue. When the size of the shared queue goes below the threshold, we are going to
// resume receiving from the paused consumer partitions
@@ -79,12 +80,12 @@ public class TopicsConsumerImpl extends ConsumerBase {
private final ReadWriteLock lock = new ReentrantReadWriteLock();
private final ConsumerStats stats;
private final UnAckedMessageTracker unAckedMessageTracker;
- private final ConsumerConfigurationData internalConfig;
+ private final ConsumerConfigurationData<T> internalConfig;
- TopicsConsumerImpl(PulsarClientImpl client, ConsumerConfigurationData conf, ExecutorService listenerExecutor,
- CompletableFuture<Consumer> subscribeFuture) {
+ TopicsConsumerImpl(PulsarClientImpl client, ConsumerConfigurationData<T> conf, ExecutorService listenerExecutor,
+ CompletableFuture<Consumer<T>> subscribeFuture, Schema<T> schema) {
super(client, "TopicsConsumerFakeTopicName" + ConsumerName.generateRandomName(), conf,
- Math.max(2, conf.getReceiverQueueSize()), listenerExecutor, subscribeFuture);
+ Math.max(2, conf.getReceiverQueueSize()), listenerExecutor, subscribeFuture, schema);
checkArgument(conf.getReceiverQueueSize() > 0,
"Receiver queue size needs to be greater than 0 for Topics Consumer");
@@ -193,7 +194,7 @@ public class TopicsConsumerImpl extends ConsumerBase {
}
}
- private void receiveMessageFromConsumer(ConsumerImpl consumer) {
+ private void receiveMessageFromConsumer(ConsumerImpl<T> consumer) {
consumer.receiveAsync().thenAccept(message -> {
if (log.isDebugEnabled()) {
log.debug("[{}] [{}] Receive message from sub consumer:{}",
@@ -238,7 +239,7 @@ public class TopicsConsumerImpl extends ConsumerBase {
// if asyncReceive is waiting : return message to callback without adding to incomingMessages queue
if (!pendingReceives.isEmpty()) {
- CompletableFuture<Message> receivedFuture = pendingReceives.poll();
+ CompletableFuture<Message<T>> receivedFuture = pendingReceives.poll();
listenerExecutor.execute(() -> receivedFuture.complete(topicMessage));
} else {
// Enqueue the message so that it can be retrieved when application calls receive()
@@ -300,8 +301,8 @@ public class TopicsConsumerImpl extends ConsumerBase {
}
@Override
- protected Message internalReceive() throws PulsarClientException {
- Message message;
+ protected Message<T> internalReceive() throws PulsarClientException {
+ Message<T> message;
try {
message = incomingMessages.take();
checkState(message instanceof TopicMessageImpl);
@@ -315,8 +316,8 @@ public class TopicsConsumerImpl extends ConsumerBase {
}
@Override
- protected Message internalReceive(int timeout, TimeUnit unit) throws PulsarClientException {
- Message message;
+ protected Message<T> internalReceive(int timeout, TimeUnit unit) throws PulsarClientException {
+ Message<T> message;
try {
message = incomingMessages.poll(timeout, unit);
if (message != null) {
@@ -332,9 +333,9 @@ public class TopicsConsumerImpl extends ConsumerBase {
}
@Override
- protected CompletableFuture<Message> internalReceiveAsync() {
- CompletableFuture<Message> result = new CompletableFuture<>();
- Message message;
+ protected CompletableFuture<Message<T>> internalReceiveAsync() {
+ CompletableFuture<Message<T>> result = new CompletableFuture<>();
+ Message<T> message;
try {
lock.writeLock().lock();
message = incomingMessages.poll(0, TimeUnit.SECONDS);
@@ -448,7 +449,7 @@ public class TopicsConsumerImpl extends ConsumerBase {
try {
if (listenerExecutor != null && !listenerExecutor.isShutdown()) {
while (!pendingReceives.isEmpty()) {
- CompletableFuture<Message> receiveFuture = pendingReceives.poll();
+ CompletableFuture<Message<T>> receiveFuture = pendingReceives.poll();
if (receiveFuture != null) {
receiveFuture.completeExceptionally(
new PulsarClientException.AlreadyClosedException("Consumer is already closed"));
@@ -484,8 +485,8 @@ public class TopicsConsumerImpl extends ConsumerBase {
return subscription;
}
- private ConsumerConfigurationData getInternalConsumerConfig() {
- ConsumerConfigurationData internalConsumerConfig = new ConsumerConfigurationData();
+ private ConsumerConfigurationData<T> getInternalConsumerConfig() {
+ ConsumerConfigurationData<T> internalConsumerConfig = new ConsumerConfigurationData<>();
internalConsumerConfig.setSubscriptionName(subscription);
internalConsumerConfig.setReceiverQueueSize(conf.getReceiverQueueSize());
internalConsumerConfig.setSubscriptionType(conf.getSubscriptionType());
@@ -640,7 +641,7 @@ public class TopicsConsumerImpl extends ConsumerBase {
log.debug("Received topic {} metadata.partitions: {}", topicName, metadata.partitions);
}
- List<CompletableFuture<Consumer>> futureList;
+ List<CompletableFuture<Consumer<T>>> futureList;
if (metadata.partitions > 1) {
this.topics.putIfAbsent(topicName, metadata.partitions);
@@ -652,9 +653,9 @@ public class TopicsConsumerImpl extends ConsumerBase {
.mapToObj(
partitionIndex -> {
String partitionName = TopicName.get(topicName).getPartition(partitionIndex).toString();
- CompletableFuture<Consumer> subFuture = new CompletableFuture<Consumer>();
- ConsumerImpl newConsumer = new ConsumerImpl(client, partitionName, internalConfig,
- client.externalExecutorProvider().getExecutor(), partitionIndex, subFuture);
+ CompletableFuture<Consumer<T>> subFuture = new CompletableFuture<>();
+ ConsumerImpl<T> newConsumer = new ConsumerImpl<>(client, partitionName, internalConfig,
+ client.externalExecutorProvider().getExecutor(), partitionIndex, subFuture, schema);
consumers.putIfAbsent(newConsumer.getTopic(), newConsumer);
return subFuture;
})
@@ -664,9 +665,9 @@ public class TopicsConsumerImpl extends ConsumerBase {
numberTopicPartitions.incrementAndGet();
partitionNumber.incrementAndGet();
- CompletableFuture<Consumer> subFuture = new CompletableFuture<Consumer>();
- ConsumerImpl newConsumer = new ConsumerImpl(client, topicName, internalConfig,
- client.externalExecutorProvider().getExecutor(), 0, subFuture);
+ CompletableFuture<Consumer<T>> subFuture = new CompletableFuture<>();
+ ConsumerImpl<T> newConsumer = new ConsumerImpl<>(client, topicName, internalConfig,
+ client.externalExecutorProvider().getExecutor(), 0, subFuture, schema);
consumers.putIfAbsent(newConsumer.getTopic(), newConsumer);
futureList = Lists.newArrayList(subFuture);
@@ -758,7 +759,7 @@ public class TopicsConsumerImpl extends ConsumerBase {
CompletableFuture<Void> unsubscribeFuture = new CompletableFuture<>();
String topicPartName = TopicName.get(topicName).getPartitionedTopicName();
- List<ConsumerImpl> consumersToUnsub = consumers.values().stream()
+ List<ConsumerImpl<T>> consumersToUnsub = consumers.values().stream()
.filter(consumer -> {
String consumerTopicName = consumer.getTopic();
if (TopicName.get(consumerTopicName).getPartitionedTopicName().equals(topicPartName)) {
@@ -769,7 +770,7 @@ public class TopicsConsumerImpl extends ConsumerBase {
}).collect(Collectors.toList());
List<CompletableFuture<Void>> futureList = consumersToUnsub.stream()
- .map(c -> c.unsubscribeAsync()).collect(Collectors.toList());
+ .map(ConsumerImpl::unsubscribeAsync).collect(Collectors.toList());
FutureUtil.waitForAll(futureList)
.whenComplete((r, ex) -> {
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java
index 72df4ae..06b6360 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ConsumerConfigurationData.java
@@ -20,10 +20,14 @@ package org.apache.pulsar.client.impl.conf;
import static com.google.common.base.Preconditions.checkArgument;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
import java.io.Serializable;
import java.util.Set;
import java.util.SortedMap;
import java.util.TreeMap;
+import lombok.Data;
import java.util.regex.Pattern;
import org.apache.pulsar.client.api.ConsumerCryptoFailureAction;
@@ -32,14 +36,8 @@ import org.apache.pulsar.client.api.CryptoKeyReader;
import org.apache.pulsar.client.api.MessageListener;
import org.apache.pulsar.client.api.SubscriptionType;
-import com.fasterxml.jackson.annotation.JsonIgnore;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
-
-import lombok.Data;
-
@Data
-public class ConsumerConfigurationData implements Serializable, Cloneable {
+public class ConsumerConfigurationData<T> implements Serializable, Cloneable {
private static final long serialVersionUID = 1L;
private Set<String> topicNames = Sets.newTreeSet();
@@ -51,7 +49,7 @@ public class ConsumerConfigurationData implements Serializable, Cloneable {
private SubscriptionType subscriptionType = SubscriptionType.Exclusive;
@JsonIgnore
- private MessageListener messageListener;
+ private MessageListener<T> messageListener;
@JsonIgnore
private ConsumerEventListener consumerEventListener;
@@ -83,7 +81,7 @@ public class ConsumerConfigurationData implements Serializable, Cloneable {
return topicNames.iterator().next();
}
- public ConsumerConfigurationData clone() {
+ public ConsumerConfigurationData<T> clone() {
try {
ConsumerConfigurationData c = (ConsumerConfigurationData) super.clone();
c.topicNames = Sets.newTreeSet(this.topicNames);
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ReaderConfigurationData.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ReaderConfigurationData.java
index d369eec..1d34669 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ReaderConfigurationData.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/conf/ReaderConfigurationData.java
@@ -28,14 +28,14 @@ import org.apache.pulsar.client.api.ReaderListener;
import lombok.Data;
@Data
-public class ReaderConfigurationData implements Serializable, Cloneable {
+public class ReaderConfigurationData<T> implements Serializable, Cloneable {
private String topicName;
private MessageId startMessageId;
private int receiverQueueSize = 1000;
- private ReaderListener readerListener;
+ private ReaderListener<T> readerListener;
private String readerName = null;
private String subscriptionRolePrefix = null;
@@ -43,9 +43,9 @@ public class ReaderConfigurationData implements Serializable, Cloneable {
private CryptoKeyReader cryptoKeyReader = null;
private ConsumerCryptoFailureAction cryptoFailureAction = ConsumerCryptoFailureAction.FAIL;
- public ReaderConfigurationData clone() {
+ public ReaderConfigurationData<T> clone() {
try {
- return (ReaderConfigurationData) super.clone();
+ return (ReaderConfigurationData<T>) super.clone();
} catch (CloneNotSupportedException e) {
throw new RuntimeException("Failed to clone ReaderConfigurationData");
}
diff --git a/pulsar-storm/src/main/java/org/apache/pulsar/storm/PulsarBolt.java b/pulsar-storm/src/main/java/org/apache/pulsar/storm/PulsarBolt.java
index bdc3c37..a293fd6 100644
--- a/pulsar-storm/src/main/java/org/apache/pulsar/storm/PulsarBolt.java
+++ b/pulsar-storm/src/main/java/org/apache/pulsar/storm/PulsarBolt.java
@@ -61,7 +61,7 @@ public class PulsarBolt extends BaseRichBolt implements IMetric {
private String componentId;
private String boltId;
private OutputCollector collector;
- private Producer producer;
+ private Producer<byte[]> producer;
private volatile long messagesSent = 0;
private volatile long messageSizeSent = 0;
@@ -107,7 +107,7 @@ public class PulsarBolt extends BaseRichBolt implements IMetric {
try {
if (producer != null) {
// a message key can be provided in the mapper
- Message msg = pulsarBoltConf.getTupleToMessageMapper().toMessage(input);
+ Message<byte[]> msg = pulsarBoltConf.getTupleToMessageMapper().toMessage(input);
if (msg == null) {
if (LOG.isDebugEnabled()) {
LOG.debug("[{}] Cannot send null message, acking the collector", boltId);
diff --git a/pulsar-storm/src/main/java/org/apache/pulsar/storm/TupleToMessageMapper.java b/pulsar-storm/src/main/java/org/apache/pulsar/storm/TupleToMessageMapper.java
index 5c464d9..7033958 100644
--- a/pulsar-storm/src/main/java/org/apache/pulsar/storm/TupleToMessageMapper.java
+++ b/pulsar-storm/src/main/java/org/apache/pulsar/storm/TupleToMessageMapper.java
@@ -33,7 +33,7 @@ public interface TupleToMessageMapper extends Serializable {
* @param tuple
* @return
*/
- public Message toMessage(Tuple tuple);
+ public Message<byte[]> toMessage(Tuple tuple);
/**
* Declare the output schema for the bolt.
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/LoadSimulationClient.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/LoadSimulationClient.java
index 54379f4..c8806ce 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/LoadSimulationClient.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/LoadSimulationClient.java
@@ -91,7 +91,7 @@ public class LoadSimulationClient {
// consumption as well as size may be changed at
// any time, and the TradeUnit may also be stopped.
private static class TradeUnit {
- Future<Consumer> consumerFuture;
+ Future<Consumer<byte[]>> consumerFuture;
final AtomicBoolean stop;
final RateLimiter rateLimiter;
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
index 920eebb..f07eeba 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceConsumer.java
@@ -246,7 +246,7 @@ public class PerformanceConsumer {
return null;
}
}
- List<Future<Consumer>> futures = Lists.newArrayList();
+ List<Future<Consumer<byte[]>>> futures = Lists.newArrayList();
ConsumerConfiguration consumerConfig = new ConsumerConfiguration();
consumerConfig.setMessageListener(listener);
consumerConfig.setReceiverQueueSize(arguments.receiverQueueSize);
@@ -275,7 +275,7 @@ public class PerformanceConsumer {
}
}
- for (Future<Consumer> future : futures) {
+ for (Future<Consumer<byte[]>> future : futures) {
future.get();
}
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
index de13421..681eb3d 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceProducer.java
@@ -231,7 +231,7 @@ public class PerformanceProducer {
// Now processing command line arguments
String prefixTopicName = arguments.topics.get(0);
- List<Future<Producer>> futures = Lists.newArrayList();
+ List<Future<Producer<byte[]>>> futures = Lists.newArrayList();
ClientConfiguration clientConf = new ClientConfiguration();
clientConf.setConnectionsPerBroker(arguments.maxConnections);
@@ -296,8 +296,8 @@ public class PerformanceProducer {
}
}
- final List<Producer> producers = Lists.newArrayListWithCapacity(futures.size());
- for (Future<Producer> future : futures) {
+ final List<Producer<byte[]>> producers = Lists.newArrayListWithCapacity(futures.size());
+ for (Future<Producer<byte[]>> future : futures) {
producers.add(future.get());
}
@@ -321,7 +321,7 @@ public class PerformanceProducer {
// Send messages on all topics/producers
long totalSent = 0;
while (true) {
- for (Producer producer : producers) {
+ for (Producer<byte[]> producer : producers) {
if (arguments.testTime > 0) {
if (System.currentTimeMillis() - startTime > arguments.testTime) {
log.info("------------------- DONE -----------------------");
diff --git a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceReader.java b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceReader.java
index de07212..c5e66b2 100644
--- a/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceReader.java
+++ b/pulsar-testclient/src/main/java/org/apache/pulsar/testclient/PerformanceReader.java
@@ -193,7 +193,7 @@ public class PerformanceReader {
clientConf.setTlsTrustCertsFilePath(arguments.tlsTrustCertsFilePath);
PulsarClient pulsarClient = new PulsarClientImpl(arguments.serviceURL, clientConf);
- List<CompletableFuture<Reader>> futures = Lists.newArrayList();
+ List<CompletableFuture<Reader<byte[]>>> futures = Lists.newArrayList();
ReaderConfiguration readerConfig = new ReaderConfiguration();
readerConfig.setReaderListener(listener);
readerConfig.setReceiverQueueSize(arguments.receiverQueueSize);
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ConsumerHandler.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ConsumerHandler.java
index 5cea30e..8b0f573 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ConsumerHandler.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ConsumerHandler.java
@@ -20,6 +20,9 @@ package org.apache.pulsar.websocket;
import static com.google.common.base.Preconditions.checkArgument;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.google.common.base.Enums;
+import com.google.common.base.Splitter;
import java.io.IOException;
import java.util.Base64;
import java.util.List;
@@ -27,10 +30,8 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.concurrent.atomic.LongAdder;
-
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
-
import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.ConsumerConfiguration;
@@ -48,10 +49,6 @@ import org.eclipse.jetty.websocket.servlet.ServletUpgradeResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.google.common.base.Enums;
-import com.google.common.base.Splitter;
-
/**
*
* WebSocket end-point url handler to handle incoming receive and acknowledge requests.
@@ -67,7 +64,7 @@ public class ConsumerHandler extends AbstractWebSocketHandler {
private String subscription = null;
private final ConsumerConfiguration conf;
- private Consumer consumer;
+ private Consumer<byte[]> consumer;
private final int maxPendingMessages;
private final AtomicInteger pendingMessages = new AtomicInteger();
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ProducerHandler.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ProducerHandler.java
index 2beb9bb..151bdbf 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ProducerHandler.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ProducerHandler.java
@@ -68,7 +68,7 @@ import com.google.common.base.Enums;
public class ProducerHandler extends AbstractWebSocketHandler {
- private Producer producer;
+ private Producer<byte[]> producer;
private final LongAdder numMsgsSent;
private final LongAdder numMsgsFailed;
private final LongAdder numBytesSent;
@@ -178,7 +178,7 @@ public class ProducerHandler extends AbstractWebSocketHandler {
if (sendRequest.replicationClusters != null) {
builder.setReplicationClusters(sendRequest.replicationClusters);
}
- Message msg = builder.build();
+ Message<byte[]> msg = builder.build();
final long now = System.nanoTime();
producer.sendAsync(msg).thenAccept(msgId -> {
diff --git a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ReaderHandler.java b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ReaderHandler.java
index c9a7970..1eddd21 100644
--- a/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ReaderHandler.java
+++ b/pulsar-websocket/src/main/java/org/apache/pulsar/websocket/ReaderHandler.java
@@ -20,18 +20,14 @@ package org.apache.pulsar.websocket;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
+import com.fasterxml.jackson.core.JsonProcessingException;
import java.io.IOException;
-import java.time.Instant;
-import java.time.ZoneId;
-import java.time.format.DateTimeFormatter;
import java.util.Base64;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import java.util.concurrent.atomic.LongAdder;
-
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
-
import org.apache.pulsar.broker.authentication.AuthenticationDataSource;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.MessageId;
@@ -50,8 +46,6 @@ import org.eclipse.jetty.websocket.servlet.ServletUpgradeResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import com.fasterxml.jackson.core.JsonProcessingException;
-
/**
*
* WebSocket end-point url handler to handle incoming receive.
@@ -63,7 +57,7 @@ import com.fasterxml.jackson.core.JsonProcessingException;
public class ReaderHandler extends AbstractWebSocketHandler {
private String subscription;
private final ReaderConfiguration conf;
- private Reader reader;
+ private Reader<byte[]> reader;
private final int maxPendingMessages;
private final AtomicInteger pendingMessages = new AtomicInteger();
--
To stop receiving notification emails like this one, please contact
mmerli@apache.org.