This is an automated email from the ASF dual-hosted git repository.
rdhabalia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 0f8aef2 [pulsar-client] Fix: set and return topic name on message api
(#11743)
0f8aef2 is described below
commit 0f8aef2b9494cde8e8adc9d97d89a65d73ae8c35
Author: Rajan Dhabalia <rdhaba...@apache.org>
AuthorDate: Sun Aug 22 13:34:34 2021 -0700
[pulsar-client] Fix: set and return topic name on message api (#11743)
---
.../api/PartitionedProducerConsumerTest.java | 40 ++++++++++++++++++++++
.../pulsar/client/impl/ProducerSemaphoreTest.java | 6 ++--
.../pulsar/client/impl/MessageBuilderImpl.java | 2 +-
.../org/apache/pulsar/client/impl/MessageImpl.java | 5 +--
.../client/impl/TypedMessageBuilderImpl.java | 2 +-
.../apache/pulsar/client/impl/MessageImplTest.java | 30 ++++++++--------
.../org/apache/pulsar/client/impl/MessageTest.java | 8 ++---
.../client/impl/schema/AutoConsumeSchemaTest.java | 2 +-
8 files changed, 68 insertions(+), 27 deletions(-)
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionedProducerConsumerTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionedProducerConsumerTest.java
index 3e9b194..543c172 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionedProducerConsumerTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/PartitionedProducerConsumerTest.java
@@ -903,6 +903,46 @@ public class PartitionedProducerConsumerTest extends
ProducerConsumerBase {
log.info("-- Exiting {} test --", methodName);
}
+ @Test
+ public void testCustomPartitionedProducer() throws Exception {
+ PulsarClient pulsarClient = newPulsarClient(lookupUrl.toString(),
0);// Creates new client connection
+ TopicName topicName = null;
+ Producer<byte[]> producer = null;
+ try {
+ log.info("-- Starting {} test --", methodName);
+
+ int numPartitions = 4;
+ topicName = TopicName
+
.get("persistent://my-property/my-ns/my-partitionedtopic1-" +
System.currentTimeMillis());
+
+ admin.topics().createPartitionedTopic(topicName.toString(),
numPartitions);
+
+ RouterWithTopicName router = new RouterWithTopicName();
+ producer = pulsarClient.newProducer().topic(topicName.toString())
+ .messageRouter(router)
+ .create();
+ for (int i = 0; i < 1; i++) {
+ String message = "my-message-" + i;
+
producer.newMessage().key(String.valueOf(i)).value(message.getBytes()).send();
+ }
+ assertEquals(router.topicName, topicName.toString());
+ } finally {
+ producer.close();
+ pulsarClient.close();
+ admin.topics().deletePartitionedTopic(topicName.toString());
+ log.info("-- Exiting {} test --", methodName);
+ }
+ }
+
+ private static class RouterWithTopicName implements MessageRouter {
+ static String topicName = null;
+
+ @Override
+ public int choosePartition(Message<?> msg, TopicMetadata metadata) {
+ topicName = msg.getTopicName();
+ return 2;
+ }
+ }
private static class AlwaysTwoMessageRouter implements MessageRouter {
@Override
diff --git
a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
index f46509e..c719cbd 100644
---
a/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
+++
b/pulsar-broker/src/test/java/org/apache/pulsar/client/impl/ProducerSemaphoreTest.java
@@ -88,7 +88,7 @@ public class ProducerSemaphoreTest extends
ProducerConsumerBase {
for (int i = 0; i < messages / 2; i++) {
MessageMetadata metadata = new MessageMetadata()
.setNumMessagesInBatch(10);
- MessageImpl<byte[]> msg = MessageImpl.create(metadata,
ByteBuffer.wrap(new byte[0]), Schema.BYTES);
+ MessageImpl<byte[]> msg = MessageImpl.create(metadata,
ByteBuffer.wrap(new byte[0]), Schema.BYTES, null);
futures.add(producer.sendAsync(msg));
}
Assert.assertEquals(producer.getSemaphore().get().availablePermits(),
pendingQueueSize - messages/2);
@@ -147,7 +147,7 @@ public class ProducerSemaphoreTest extends
ProducerConsumerBase {
MessageMetadata metadata = new MessageMetadata()
.setNumMessagesInBatch(10);
- MessageImpl<byte[]> msg = MessageImpl.create(metadata,
ByteBuffer.wrap(new byte[0]), Schema.BYTES);
+ MessageImpl<byte[]> msg = MessageImpl.create(metadata,
ByteBuffer.wrap(new byte[0]), Schema.BYTES, null);
futures.add(producer.sendAsync(msg));
}
Assert.assertEquals(producer.getSemaphore().get().availablePermits(), 0);
@@ -155,7 +155,7 @@ public class ProducerSemaphoreTest extends
ProducerConsumerBase {
MessageMetadata metadata = new MessageMetadata()
.setNumMessagesInBatch(10);
- MessageImpl<byte[]> msg = MessageImpl.create(metadata,
ByteBuffer.wrap(new byte[0]), Schema.BYTES);
+ MessageImpl<byte[]> msg = MessageImpl.create(metadata,
ByteBuffer.wrap(new byte[0]), Schema.BYTES, null);
producer.sendAsync(msg).get();
Assert.fail("Shouldn't be able to send message");
} catch (ExecutionException ee) {
diff --git
a/pulsar-client-1x-base/pulsar-client-1x/src/main/java/org/apache/pulsar/client/impl/MessageBuilderImpl.java
b/pulsar-client-1x-base/pulsar-client-1x/src/main/java/org/apache/pulsar/client/impl/MessageBuilderImpl.java
index 103d322..11c7dfa 100644
---
a/pulsar-client-1x-base/pulsar-client-1x/src/main/java/org/apache/pulsar/client/impl/MessageBuilderImpl.java
+++
b/pulsar-client-1x-base/pulsar-client-1x/src/main/java/org/apache/pulsar/client/impl/MessageBuilderImpl.java
@@ -40,7 +40,7 @@ public class MessageBuilderImpl implements MessageBuilder {
@Override
public Message<byte[]> build() {
- return MessageImpl.create(msgMetadataBuilder, content, Schema.BYTES);
+ return MessageImpl.create(msgMetadataBuilder, content, Schema.BYTES,
null);
}
@Override
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 0297b14..fef98e7 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
@@ -78,13 +78,14 @@ public class MessageImpl<T> implements Message<T> {
private boolean poolMessage;
// Constructor for out-going message
- public static <T> MessageImpl<T> create(MessageMetadata msgMetadata,
ByteBuffer payload, Schema<T> schema) {
+ public static <T> MessageImpl<T> create(MessageMetadata msgMetadata,
ByteBuffer payload, Schema<T> schema,
+ String topic) {
@SuppressWarnings("unchecked")
MessageImpl<T> msg = (MessageImpl<T>) RECYCLER.get();
msg.msgMetadata.clear();
msg.msgMetadata.copyFrom(msgMetadata);
msg.messageId = null;
- msg.topic = null;
+ msg.topic = topic;
msg.cnx = null;
msg.payload = Unpooled.wrappedBuffer(payload);
msg.properties = null;
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TypedMessageBuilderImpl.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TypedMessageBuilderImpl.java
index 51dc992..f981bc5 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TypedMessageBuilderImpl.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/TypedMessageBuilderImpl.java
@@ -284,7 +284,7 @@ public class TypedMessageBuilderImpl<T> implements
TypedMessageBuilder<T> {
public Message<T> getMessage() {
beforeSend();
- return MessageImpl.create(msgMetadata, content, schema);
+ return MessageImpl.create(msgMetadata, content, schema,
producer.topic);
}
public long getPublishTime() {
diff --git
a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MessageImplTest.java
b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MessageImplTest.java
index 4083c34..1580570 100644
---
a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MessageImplTest.java
+++
b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MessageImplTest.java
@@ -59,7 +59,7 @@ public class MessageImplTest {
@Test
public void testGetSequenceIdNotAssociated() {
ByteBuffer payload = ByteBuffer.wrap(new byte[0]);
- MessageImpl<?> msg = MessageImpl.create(new MessageMetadata(),
payload, Schema.BYTES);
+ MessageImpl<?> msg = MessageImpl.create(new MessageMetadata(),
payload, Schema.BYTES, null);
assertEquals(-1, msg.getSequenceId());
}
@@ -71,7 +71,7 @@ public class MessageImplTest {
builder.addProperty().setKey("key1").setValue("value2");
builder.addProperty().setKey("key3").setValue("value3");
ByteBuffer payload = ByteBuffer.wrap(new byte[0]);
- MessageImpl<?> msg = MessageImpl.create(builder, payload,
Schema.BYTES);
+ MessageImpl<?> msg = MessageImpl.create(builder, payload,
Schema.BYTES, null);
assertEquals("value2", msg.getProperty("key1"));
assertEquals("value3", msg.getProperty("key3"));
}
@@ -82,7 +82,7 @@ public class MessageImplTest {
.setSequenceId(1234);
ByteBuffer payload = ByteBuffer.wrap(new byte[0]);
- MessageImpl<?> msg = MessageImpl.create(builder, payload,
Schema.BYTES);
+ MessageImpl<?> msg = MessageImpl.create(builder, payload,
Schema.BYTES, null);
assertEquals(1234, msg.getSequenceId());
}
@@ -91,7 +91,7 @@ public class MessageImplTest {
public void testGetProducerNameNotAssigned() {
MessageMetadata builder = new MessageMetadata();
ByteBuffer payload = ByteBuffer.wrap(new byte[0]);
- MessageImpl<?> msg = MessageImpl.create(builder, payload,
Schema.BYTES);
+ MessageImpl<?> msg = MessageImpl.create(builder, payload,
Schema.BYTES, null);
assertNull(msg.getProducerName());
}
@@ -102,7 +102,7 @@ public class MessageImplTest {
.setProducerName("test-producer");
ByteBuffer payload = ByteBuffer.wrap(new byte[0]);
- MessageImpl<?> msg = MessageImpl.create(builder, payload,
Schema.BYTES);
+ MessageImpl<?> msg = MessageImpl.create(builder, payload,
Schema.BYTES, null);
assertEquals("test-producer", msg.getProducerName());
}
@@ -127,7 +127,7 @@ public class MessageImplTest {
MessageMetadata builder = new MessageMetadata()
.setProducerName("default");
MessageImpl<KeyValue<SchemaTestUtils.Foo, SchemaTestUtils.Bar>> msg =
MessageImpl.create(
- builder, ByteBuffer.wrap(encodeBytes), keyValueSchema);
+ builder, ByteBuffer.wrap(encodeBytes), keyValueSchema, null);
KeyValue<SchemaTestUtils.Foo, SchemaTestUtils.Bar> keyValue =
msg.getValue();
assertEquals(keyValue.getKey(), foo);
assertEquals(keyValue.getValue(), bar);
@@ -156,7 +156,7 @@ public class MessageImplTest {
MessageMetadata builder = new MessageMetadata()
.setProducerName("inline");
MessageImpl<KeyValue<SchemaTestUtils.Foo, SchemaTestUtils.Bar>> msg =
MessageImpl.create(
- builder, ByteBuffer.wrap(encodeBytes), keyValueSchema);
+ builder, ByteBuffer.wrap(encodeBytes), keyValueSchema, null);
KeyValue<SchemaTestUtils.Foo, SchemaTestUtils.Bar> keyValue =
msg.getValue();
assertEquals(keyValue.getKey(), foo);
assertEquals(keyValue.getValue(), bar);
@@ -186,7 +186,7 @@ public class MessageImplTest {
builder.setPartitionKey(Base64.getEncoder().encodeToString(fooSchema.encode(foo)));
builder.setPartitionKeyB64Encoded(true);
MessageImpl<KeyValue<SchemaTestUtils.Foo, SchemaTestUtils.Bar>> msg =
MessageImpl.create(
- builder, ByteBuffer.wrap(encodeBytes), keyValueSchema);
+ builder, ByteBuffer.wrap(encodeBytes), keyValueSchema, null);
KeyValue<SchemaTestUtils.Foo, SchemaTestUtils.Bar> keyValue =
msg.getValue();
assertEquals(keyValue.getKey(), foo);
assertEquals(keyValue.getValue(), bar);
@@ -218,7 +218,7 @@ public class MessageImplTest {
.setProducerName("default");
builder.setSchemaVersion(new byte[10]);
MessageImpl<KeyValue<SchemaTestUtils.Foo, SchemaTestUtils.Bar>> msg =
MessageImpl.create(
- builder, ByteBuffer.wrap(encodeBytes), keyValueSchema);
+ builder, ByteBuffer.wrap(encodeBytes), keyValueSchema, null);
KeyValue<SchemaTestUtils.Foo, SchemaTestUtils.Bar> keyValue =
msg.getValue();
assertEquals(keyValue.getKey(), foo);
assertEquals(keyValue.getValue(), bar);
@@ -256,7 +256,7 @@ public class MessageImplTest {
builder.setPartitionKey(Base64.getEncoder().encodeToString(fooSchema.encode(foo)));
builder.setPartitionKeyB64Encoded(true);
MessageImpl<KeyValue<SchemaTestUtils.Foo, SchemaTestUtils.Bar>> msg =
MessageImpl.create(
- builder, ByteBuffer.wrap(encodeBytes), keyValueSchema);
+ builder, ByteBuffer.wrap(encodeBytes), keyValueSchema, null);
KeyValue<SchemaTestUtils.Foo, SchemaTestUtils.Bar> keyValue =
msg.getValue();
assertEquals(keyValue.getKey(), foo);
assertEquals(keyValue.getValue(), bar);
@@ -291,7 +291,7 @@ public class MessageImplTest {
.setProducerName("default");
builder.setSchemaVersion(new byte[10]);
MessageImpl<KeyValue<SchemaTestUtils.Foo, SchemaTestUtils.Bar>> msg =
MessageImpl.create(
- builder, ByteBuffer.wrap(encodeBytes), keyValueSchema);
+ builder, ByteBuffer.wrap(encodeBytes), keyValueSchema, null);
KeyValue<SchemaTestUtils.Foo, SchemaTestUtils.Bar> keyValue =
msg.getValue();
assertEquals(keyValue.getKey(), foo);
assertEquals(keyValue.getValue(), bar);
@@ -329,7 +329,7 @@ public class MessageImplTest {
builder.setPartitionKey(Base64.getEncoder().encodeToString(fooSchema.encode(foo)));
builder.setPartitionKeyB64Encoded(true);
MessageImpl<KeyValue<SchemaTestUtils.Foo, SchemaTestUtils.Bar>> msg =
MessageImpl.create(
- builder, ByteBuffer.wrap(encodeBytes), keyValueSchema);
+ builder, ByteBuffer.wrap(encodeBytes), keyValueSchema, null);
KeyValue<SchemaTestUtils.Foo, SchemaTestUtils.Bar> keyValue =
msg.getValue();
assertEquals(keyValue.getKey(), foo);
assertEquals(keyValue.getValue(), bar);
@@ -364,7 +364,7 @@ public class MessageImplTest {
.setProducerName("default");
builder.setSchemaVersion(new byte[10]);
MessageImpl<KeyValue<SchemaTestUtils.Foo, SchemaTestUtils.Bar>> msg =
MessageImpl.create(
- builder, ByteBuffer.wrap(encodeBytes), keyValueSchema);
+ builder, ByteBuffer.wrap(encodeBytes), keyValueSchema, null);
KeyValue<SchemaTestUtils.Foo, SchemaTestUtils.Bar> keyValue =
msg.getValue();
assertEquals(keyValue.getKey(), foo);
assertEquals(keyValue.getValue(), bar);
@@ -402,7 +402,7 @@ public class MessageImplTest {
builder.setPartitionKey(Base64.getEncoder().encodeToString(fooSchema.encode(foo)));
builder.setPartitionKeyB64Encoded(true);
MessageImpl<KeyValue<SchemaTestUtils.Foo, SchemaTestUtils.Bar>> msg =
MessageImpl.create(
- builder, ByteBuffer.wrap(encodeBytes), keyValueSchema);
+ builder, ByteBuffer.wrap(encodeBytes), keyValueSchema, null);
KeyValue<SchemaTestUtils.Foo, SchemaTestUtils.Bar> keyValue =
msg.getValue();
assertEquals(keyValue.getKey(), foo);
assertEquals(keyValue.getValue(), bar);
@@ -421,7 +421,7 @@ public class MessageImplTest {
builder.setPartitionKey(Base64.getEncoder().encodeToString(encodeBytes));
builder.setPartitionKeyB64Encoded(true);
builder.setNullValue(true);
- MessageImpl<Boolean> msg = MessageImpl.create(builder,
ByteBuffer.wrap(encodeBytes), BooleanSchema.of());
+ MessageImpl<Boolean> msg = MessageImpl.create(builder,
ByteBuffer.wrap(encodeBytes), BooleanSchema.of(), null);
assertNull(msg.getValue());
}
diff --git
a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MessageTest.java
b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MessageTest.java
index 15a4f99..6d633e7 100644
--- a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MessageTest.java
+++ b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/MessageTest.java
@@ -39,7 +39,7 @@ public class MessageTest {
String from = "ClusterNameOfReplicatedFrom";
MessageMetadata builder = new
MessageMetadata().setReplicatedFrom(from);
ByteBuffer payload = ByteBuffer.wrap(new byte[0]);
- Message<byte[]> msg = MessageImpl.create(builder, payload,
Schema.BYTES);
+ Message<byte[]> msg = MessageImpl.create(builder, payload,
Schema.BYTES, null);
assertTrue(msg.isReplicated());
assertEquals(msg.getReplicatedFrom(), from);
@@ -49,7 +49,7 @@ public class MessageTest {
public void testMessageImplNoReplicatedInfo() {
MessageMetadata builder = new MessageMetadata();
ByteBuffer payload = ByteBuffer.wrap(new byte[0]);
- Message<byte[]> msg = MessageImpl.create(builder, payload,
Schema.BYTES);
+ Message<byte[]> msg = MessageImpl.create(builder, payload,
Schema.BYTES, null);
assertFalse(msg.isReplicated());
assertNull(msg.getReplicatedFrom());
@@ -61,7 +61,7 @@ public class MessageTest {
String topicName = "myTopic";
MessageMetadata builder = new
MessageMetadata().setReplicatedFrom(from);
ByteBuffer payload = ByteBuffer.wrap(new byte[0]);
- MessageImpl<byte[]> msg = MessageImpl.create(builder, payload,
Schema.BYTES);
+ MessageImpl<byte[]> msg = MessageImpl.create(builder, payload,
Schema.BYTES, null);
msg.setMessageId(new MessageIdImpl(-1, -1, -1));
TopicMessageImpl<byte[]> topicMessage = new
TopicMessageImpl<>(topicName, topicName, msg, null);
@@ -74,7 +74,7 @@ public class MessageTest {
String topicName = "myTopic";
MessageMetadata builder = new MessageMetadata();
ByteBuffer payload = ByteBuffer.wrap(new byte[0]);
- MessageImpl<byte[]> msg = MessageImpl.create(builder, payload,
Schema.BYTES);
+ MessageImpl<byte[]> msg = MessageImpl.create(builder, payload,
Schema.BYTES, null);
msg.setMessageId(new MessageIdImpl(-1, -1, -1));
TopicMessageImpl<byte[]> topicMessage = new
TopicMessageImpl<>(topicName, topicName, msg, null);
diff --git
a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/AutoConsumeSchemaTest.java
b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/AutoConsumeSchemaTest.java
index fce2ef1..646f657 100644
---
a/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/AutoConsumeSchemaTest.java
+++
b/pulsar-client/src/test/java/org/apache/pulsar/client/impl/schema/AutoConsumeSchemaTest.java
@@ -38,7 +38,7 @@ public class AutoConsumeSchemaTest {
Schema<GenericRecord> autoConsumeSchema = new AutoConsumeSchema();
byte[] bytes = "bytes data".getBytes();
MessageImpl<GenericRecord> message = MessageImpl.create(
- new MessageMetadata(), ByteBuffer.wrap(bytes),
autoConsumeSchema);
+ new MessageMetadata(), ByteBuffer.wrap(bytes),
autoConsumeSchema, null);
Assert.assertNull(message.getSchemaVersion());
GenericRecord genericRecord = message.getValue();
Assert.assertEquals(genericRecord.getNativeObject(), bytes);
