Skip to content

[PIP-43] Support producer to send msg with different schema #5165

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
sijie merged 2 commits into from
Oct 25, 2019
Merged

[PIP-43] Support producer to send msg with different schema #5165

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e12e435
support producer to send msg with different schema
yittg Sep 9, 2019
8d5e8c8
executing registering schema asynchronously
yittg Sep 20, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 55 additions & 17 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,7 @@
import org.apache.pulsar.common.api.proto.PulsarApi.CommandFlow;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandGetLastMessageId;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandGetSchema;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandGetOrCreateSchema;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandGetTopicsOfNamespace;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandLookupTopic;
import org.apache.pulsar.common.api.proto.PulsarApi.CommandPartitionedTopicMetadata;
Expand All @@ -103,6 +104,7 @@
import org.apache.pulsar.common.protocol.schema.SchemaData;
import org.apache.pulsar.common.protocol.schema.SchemaInfoUtil;
import org.apache.pulsar.common.protocol.schema.SchemaVersion;
import org.apache.pulsar.common.schema.SchemaType;
import org.apache.pulsar.common.util.FutureUtil;
import org.apache.pulsar.common.util.collections.ConcurrentLongHashMap;
import org.apache.pulsar.shaded.com.google.protobuf.v241.GeneratedMessageLite;
Expand Down Expand Up @@ -908,23 +910,7 @@ protected void handleProducer(final CommandProducer cmdProducer) {

disableTcpNoDelayIfNeeded(topicName.toString(), producerName);

CompletableFuture<SchemaVersion> schemaVersionFuture;
if (schema != null) {
schemaVersionFuture = topic.addSchema(schema);
} else {
schemaVersionFuture = topic.hasSchema().thenCompose((hasSchema) -> {
log.info("[{}]-{} {} configured with schema {}", remoteAddress, producerId,
topicName, hasSchema);
CompletableFuture<SchemaVersion> result = new CompletableFuture<>();
if (hasSchema && (schemaValidationEnforced || topic.getSchemaValidationEnforced())) {
result.completeExceptionally(new IncompatibleSchemaException(
"Producers cannot connect without a schema to topics with a schema"));
} else {
result.complete(SchemaVersion.Empty);
}
return result;
});
}
CompletableFuture<SchemaVersion> schemaVersionFuture = tryAddSchema(topic, schema);

schemaVersionFuture.exceptionally(exception -> {
ctx.writeAndFlush(Commands.newError(requestId,
Expand Down Expand Up @@ -1356,6 +1342,58 @@ protected void handleGetSchema(CommandGetSchema commandGetSchema) {
});
}

@Override
protected void handleGetOrCreateSchema(CommandGetOrCreateSchema commandGetOrCreateSchema) {
if (log.isDebugEnabled()) {
log.debug("Received CommandGetOrCreateSchema call from {}", remoteAddress);
}
long requestId = commandGetOrCreateSchema.getRequestId();
String topicName = commandGetOrCreateSchema.getTopic();
SchemaData schemaData = getSchema(commandGetOrCreateSchema.getSchema());
SchemaData schema = schemaData.getType() == SchemaType.NONE ? null : schemaData;
service.getTopicIfExists(topicName).thenAccept(topicOpt -> {
if (topicOpt.isPresent()) {
Topic topic = topicOpt.get();
CompletableFuture<SchemaVersion> schemaVersionFuture = tryAddSchema(topic, schema);
schemaVersionFuture.exceptionally(ex -> {
ServerError errorCode = BrokerServiceException.getClientErrorCode(ex);
ctx.writeAndFlush(Commands.newGetOrCreateSchemaResponseError(
requestId, errorCode, ex.getMessage()));
return null;
}).thenAccept(schemaVersion -> {
ctx.writeAndFlush(Commands.newGetOrCreateSchemaResponse(
requestId, schemaVersion));
});
} else {
ctx.writeAndFlush(Commands.newGetOrCreateSchemaResponseError(
requestId, ServerError.TopicNotFound, "Topic not found"));
}
}).exceptionally(ex -> {
ServerError errorCode = BrokerServiceException.getClientErrorCode(ex);
ctx.writeAndFlush(Commands.newGetOrCreateSchemaResponseError(
requestId, errorCode, ex.getMessage()));
return null;
});
}

private CompletableFuture<SchemaVersion> tryAddSchema(Topic topic, SchemaData schema) {
if (schema != null) {
return topic.addSchema(schema);
} else {
return topic.hasSchema().thenCompose((hasSchema) -> {
log.info("[{}] {} configured with schema {}",

@sijie sijie Oct 14, 2019

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we keep the origin log statement when refactoring the code?

@yittg yittg Oct 14, 2019

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

producer_id not carried in CommandGetOrCreateSchema, actually need it?
Or a -1 as placeholder?

remoteAddress, topic.getName(), hasSchema);
CompletableFuture<SchemaVersion> result = new CompletableFuture<>();
if (hasSchema && (schemaValidationEnforced || topic.getSchemaValidationEnforced())) {
result.completeExceptionally(new IncompatibleSchemaException(
"Producers cannot connect or send message without a schema to topics with a schema"));
} else {
result.complete(SchemaVersion.Empty);
}
return result;
});
}
}

@Override
protected boolean isHandshakeCompleted() {
Expand Down
220 changes: 204 additions & 16 deletions pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleSchemaTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,22 +18,40 @@
*/
package org.apache.pulsar.client.api;

import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;

import static java.nio.charset.StandardCharsets.UTF_8;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;

import org.apache.avro.reflect.ReflectData;
import org.apache.avro.Schema.Parser;
import org.apache.pulsar.broker.service.schema.LongSchemaVersion;
import org.apache.pulsar.client.api.PulsarClientException.IncompatibleSchemaException;
import org.apache.pulsar.client.api.PulsarClientException.InvalidMessageException;
import org.apache.pulsar.client.api.schema.GenericRecord;
import org.apache.pulsar.client.impl.ProducerBase;
import org.apache.pulsar.client.impl.schema.writer.AvroWriter;
import org.apache.pulsar.common.protocol.schema.SchemaVersion;
import org.apache.pulsar.common.schema.KeyValue;
import org.apache.pulsar.common.schema.KeyValueEncodingType;
import org.apache.pulsar.common.schema.SchemaInfo;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Factory;
import org.testng.annotations.Test;

import java.io.ByteArrayInputStream;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class SimpleSchemaTest extends ProducerConsumerBase {

@DataProvider(name = "batchingModes")
Expand Down Expand Up @@ -94,26 +112,30 @@ public void testString() throws Exception {
}
}

@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
static class V1Data {
int i;
}

V1Data() {
this.i = 0;
}

V1Data(int i) {
this.i = i;
}

@Override
public int hashCode() {
return i;
}
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
static class V2Data {
int i;
Integer j;
}

@Override
public boolean equals(Object other) {
return (other instanceof V1Data) && i == ((V1Data)other).i;
}
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
static class IncompatibleData {
int i;
int j;
}

@Test
Expand Down Expand Up @@ -185,6 +207,172 @@ public void newProducerWithoutSchemaOnTopicWithSchema() throws Exception {
}
}

@Test
public void newProducerForMessageSchemaOnTopicWithMultiVersionSchema() throws Exception {
String topic = "my-property/my-ns/schema-test";
Schema<V1Data> v1Schema = Schema.AVRO(V1Data.class);
byte[] v1SchemaBytes = v1Schema.getSchemaInfo().getSchema();
AvroWriter<V1Data> v1Writer = new AvroWriter<>(
new Parser().parse(new ByteArrayInputStream(v1SchemaBytes)));
Schema<V2Data> v2Schema = Schema.AVRO(V2Data.class);
byte[] v2SchemaBytes = v2Schema.getSchemaInfo().getSchema();
AvroWriter<V2Data> v2Writer = new AvroWriter<>(
new Parser().parse(new ByteArrayInputStream(v2SchemaBytes)));
try (Producer<V1Data> ignored = pulsarClient.newProducer(v1Schema)
.topic(topic).create()) {
}
try (Producer<V2Data> p = pulsarClient.newProducer(Schema.AVRO(V2Data.class))
.topic(topic).create()) {
p.send(new V2Data(-1, -1));
}
V1Data dataV1 = new V1Data(2);
V2Data dataV2 = new V2Data(3, 5);
byte[] contentV1 = v1Writer.write(dataV1);
byte[] contentV2 = v2Writer.write(dataV2);
try (Producer<byte[]> p = pulsarClient.newProducer(Schema.AUTO_PRODUCE_BYTES())
.topic(topic).create();
Consumer<V2Data> c = pulsarClient.newConsumer(v2Schema)
.topic(topic)
.subscriptionName("sub1").subscribe()) {
Assert.expectThrows(SchemaSerializationException.class, () -> p.send(contentV1));

((ProducerBase<byte[]>)p).newMessage(Schema.AUTO_PRODUCE_BYTES(Schema.AVRO(V1Data.class)))
.value(contentV1).send();
p.send(contentV2);
Message<V2Data> msg1 = c.receive();
V2Data msg1Value = msg1.getValue();
Assert.assertEquals(dataV1.i, msg1Value.i);
Assert.assertNull(msg1Value.j);
Assert.assertEquals(msg1.getSchemaVersion(), new LongSchemaVersion(0).bytes());

Message<V2Data> msg2 = c.receive();
Assert.assertEquals(dataV2, msg2.getValue());
Assert.assertEquals(msg2.getSchemaVersion(), new LongSchemaVersion(1).bytes());

try {
((ProducerBase<byte[]>)p).newMessage(Schema.BYTES).value(contentV1).send();
if (schemaValidationEnforced) {
Assert.fail("Shouldn't be able to send to a schema'd topic with no schema"
+ " if SchemaValidationEnabled is enabled");
}
Message<V2Data> msg3 = c.receive();
Assert.assertEquals(msg3.getSchemaVersion(), SchemaVersion.Empty.bytes());
} catch (PulsarClientException e) {
if (schemaValidationEnforced) {
Assert.assertTrue(e instanceof IncompatibleSchemaException);
} else {
Assert.fail("Shouldn't throw IncompatibleSchemaException"
+ " if SchemaValidationEnforced is disabled");
}
}
}
}

@Test
public void newProducerForMessageSchemaOnTopicInitialWithNoSchema() throws Exception {
String topic = "my-property/my-ns/schema-test";
Schema<V1Data> v1Schema = Schema.AVRO(V1Data.class);
byte[] v1SchemaBytes = v1Schema.getSchemaInfo().getSchema();
AvroWriter<V1Data> v1Writer = new AvroWriter<>(
new Parser().parse(new ByteArrayInputStream(v1SchemaBytes)));
Schema<V2Data> v2Schema = Schema.AVRO(V2Data.class);
byte[] v2SchemaBytes = v2Schema.getSchemaInfo().getSchema();
AvroWriter<V2Data> v2Writer = new AvroWriter<>(
new Parser().parse(new ByteArrayInputStream(v2SchemaBytes)));
try (Producer<byte[]> p = pulsarClient.newProducer()
.topic(topic).create();
Consumer<byte[]> c = pulsarClient.newConsumer()
.topic(topic)
.subscriptionName("sub1").subscribe()) {
for (int i = 0; i < 2; ++i) {
V1Data dataV1 = new V1Data(i);
V2Data dataV2 = new V2Data(i, -i);
byte[] contentV1 = v1Writer.write(dataV1);
byte[] contentV2 = v2Writer.write(dataV2);
((ProducerBase<byte[]>) p).newMessage(Schema.AUTO_PRODUCE_BYTES(v1Schema))
.value(contentV1).send();
Message<byte[]> msg1 = c.receive();
Assert.assertEquals(msg1.getSchemaVersion(), new LongSchemaVersion(0).bytes());
Assert.assertEquals(msg1.getData(), contentV1);
((ProducerBase<byte[]>) p).newMessage(Schema.AUTO_PRODUCE_BYTES(v2Schema))
.value(contentV2).send();
Message<byte[]> msg2 = c.receive();
Assert.assertEquals(msg2.getSchemaVersion(), new LongSchemaVersion(1).bytes());
Assert.assertEquals(msg2.getData(), contentV2);
}
}

List<SchemaInfo> allSchemas = admin.schemas().getAllSchemas(topic);
Assert.assertEquals(allSchemas, Arrays.asList(v1Schema.getSchemaInfo(),
v2Schema.getSchemaInfo()));
}

@Test
public void newProducerForMessageSchemaWithBatch() throws Exception {
String topic = "my-property/my-ns/schema-test";
Consumer<V2Data> c = pulsarClient.newConsumer(Schema.AVRO(V2Data.class))
.topic(topic)
.subscriptionName("sub1").subscribe();
Producer<byte[]> p = pulsarClient.newProducer(Schema.AUTO_PRODUCE_BYTES())
.topic(topic)
.enableBatching(true)
.batchingMaxPublishDelay(10, TimeUnit.SECONDS).create();
AvroWriter<V1Data> v1DataAvroWriter = new AvroWriter<>(
ReflectData.AllowNull.get().getSchema(V1Data.class));
AvroWriter<V2Data> v2DataAvroWriter = new AvroWriter<>(
ReflectData.AllowNull.get().getSchema(V2Data.class));
AvroWriter<IncompatibleData> incompatibleDataAvroWriter = new AvroWriter<>(
ReflectData.AllowNull.get().getSchema(IncompatibleData.class));
int total = 20;
int batch = 5;
int incompatible = 3;
for (int i = 0; i < total; ++i) {
if (i / batch % 2 == 0) {
byte[] content = v1DataAvroWriter.write(new V1Data(i));
((ProducerBase<byte[]>)p).newMessage(Schema.AUTO_PRODUCE_BYTES(Schema.AVRO(V1Data.class)))
.value(content).sendAsync();
} else {
byte[] content = v2DataAvroWriter.write(new V2Data(i, i + total));
((ProducerBase<byte[]>)p).newMessage(Schema.AUTO_PRODUCE_BYTES(Schema.AVRO(V2Data.class)))
.value(content).sendAsync();
}
if ((i + 1) % incompatible == 0) {
byte[] content = incompatibleDataAvroWriter.write(new IncompatibleData(-i, -i));
try {
((ProducerBase<byte[]>)p).newMessage(Schema.AUTO_PRODUCE_BYTES(Schema.AVRO(IncompatibleData.class)))
.value(content).send();
} catch (Exception e) {
Assert.assertTrue(e instanceof IncompatibleSchemaException, e.getMessage());
}
}
}
p.flush();
for (int i = 0; i < total; ++i) {
V2Data value = c.receive().getValue();
if (i / batch % 2 == 0) {
Assert.assertNull(value.j);
Assert.assertEquals(value.i, i);
} else {
Assert.assertEquals(value, new V2Data(i, i + total));
}
}
c.close();
}

@Test
public void newProducerWithMultipleSchemaDisabled() throws Exception {
String topic = "my-property/my-ns/schema-test";
AvroWriter<V1Data> v1DataAvroWriter = new AvroWriter<>(
ReflectData.AllowNull.get().getSchema(V1Data.class));
try (Producer<byte[]> p = pulsarClient.newProducer()
.topic(topic)
.enableMultiSchema(false).create()) {
Assert.assertThrows(InvalidMessageException.class,
() -> ((ProducerBase<byte[]>)p).newMessage(Schema.AUTO_PRODUCE_BYTES(Schema.AVRO(V1Data.class)))
.value(v1DataAvroWriter.write(new V1Data(0))).send());
}
}

@Test
public void newConsumerWithSchemaOnNewTopic() throws Exception {
String topic = "my-property/my-ns/schema-test";
Expand Down
15 changes: 15 additions & 0 deletions pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ProducerBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -431,4 +431,19 @@ public interface ProducerBuilder<T> extends Cloneable {
* @return the producer builder instance
*/
ProducerBuilder<T> autoUpdatePartitions(boolean autoUpdate);

/**
* Control whether enable the multiple schema mode for producer.
* If enabled, producer can send a message with different schema from that specified just when it is created,
* otherwise a invalid message exception would be threw
* if the producer want to send a message with different schema.
*
* <p>Enabled by default.
*
* @param multiSchema
* indicates to enable or disable multiple schema mode
* @return the producer builder instance
* @since 2.5.0
*/
ProducerBuilder<T> enableMultiSchema(boolean multiSchema);
}
12 changes: 12 additions & 0 deletions pulsar-client-api/src/main/java/org/apache/pulsar/client/api/Schema.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -333,6 +333,18 @@ static Schema<byte[]> AUTO_PRODUCE_BYTES() {
return DefaultImplementation.newAutoProduceSchema();
}

/**
* Create a schema instance that accepts a serialized payload
* and validates it against the schema specified.
*
* @return the auto schema instance
* @since 2.5.0
* @see #AUTO_PRODUCE_BYTES()
*/
static Schema<byte[]> AUTO_PRODUCE_BYTES(Schema<?> schema) {
return DefaultImplementation.newAutoProduceSchema(schema);
}

// CHECKSTYLE.ON: MethodName

static Schema<?> getSchema(SchemaInfo schemaInfo) {
Expand Down
Loading