Request a sample program for PIP-43

[guofengzh]

After searching for PIP 43: producer send message with different schema on Google, I found the SimpleSchemaTest class in this project for testing PIP 43. However, V1Data and V2Data used in the test are compatible.

I need entirely different business schemas (different Java classes, like UserCreated and OrderPlaced) sending to the same topic. Where can I find sample programs like this?

Thanks for your help very much.

[lhotari]

Example of producing different schemas to the same topic and consuming them:

pulsar/pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java

Lines 1168 to 1227 in 67dafa1

	@Test
	public void testConsumeMultipleSchemaMessages() throws Exception {
	final String namespace = "test-namespace-" + randomName(16);
	String ns = PUBLIC_TENANT + "/" + namespace;
	admin.namespaces().createNamespace(ns, Sets.newHashSet(CLUSTER_NAME));
	admin.namespaces().setSchemaCompatibilityStrategy(ns, SchemaCompatibilityStrategy.ALWAYS_COMPATIBLE);
	final String autoProducerTopic = getTopicName(ns, "auto_produce_topic");
	Producer<byte[]> autoProducer = pulsarClient.newProducer(Schema.AUTO_PRODUCE_BYTES())
	.topic(autoProducerTopic)
	.create();
	AtomicInteger totalMsgCnt = new AtomicInteger(0);
	generateDataByDifferentSchema(ns, "bytes_schema", Schema.BYTES, "bytes value".getBytes(),
	autoProducer, totalMsgCnt);
	generateDataByDifferentSchema(ns, "string_schema", Schema.STRING, "string value",
	autoProducer, totalMsgCnt);
	generateDataByDifferentSchema(ns, "bool_schema", Schema.BOOL, true,
	autoProducer, totalMsgCnt);
	generateDataByDifferentSchema(ns, "json_one_schema", Schema.JSON(Schemas.PersonOne.class),
	new Schemas.PersonOne(1), autoProducer, totalMsgCnt);
	generateDataByDifferentSchema(ns, "json_three_schema", Schema.JSON(Schemas.PersonThree.class),
	new Schemas.PersonThree(3, "ran"), autoProducer, totalMsgCnt);
	generateDataByDifferentSchema(ns, "json_four_schema", Schema.JSON(Schemas.PersonFour.class),
	new Schemas.PersonFour(4, "tang", 18), autoProducer, totalMsgCnt);
	generateDataByDifferentSchema(ns, "avro_one_schema", Schema.AVRO(Schemas.PersonOne.class),
	new Schemas.PersonOne(10), autoProducer, totalMsgCnt);
	generateDataByDifferentSchema(ns, "k_one_v_three_schema_separate",
	Schema.KeyValue(Schema.JSON(Schemas.PersonOne.class),
	Schema.JSON(Schemas.PersonThree.class), KeyValueEncodingType.SEPARATED),
	new KeyValue<>(new Schemas.PersonOne(1), new Schemas.PersonThree(3, "kv-separate")),
	autoProducer, totalMsgCnt);
	generateDataByDifferentSchema(ns, "k_one_v_four_schema_inline",
	Schema.KeyValue(Schema.JSON(Schemas.PersonOne.class),
	Schema.JSON(Schemas.PersonFour.class), KeyValueEncodingType.INLINE),
	new KeyValue<>(new Schemas.PersonOne(10), new Schemas.PersonFour(30, "kv-inline", 20)),
	autoProducer, totalMsgCnt);
	generateDataByDifferentSchema(ns, "k_int_v_three_schema_separate",
	Schema.KeyValue(Schema.INT32, Schema.JSON(Schemas.PersonThree.class),
	KeyValueEncodingType.SEPARATED),
	new KeyValue<>(100, new Schemas.PersonThree(40, "kv-separate")),
	autoProducer, totalMsgCnt);
	Consumer<GenericRecord> autoConsumer = pulsarClient.newConsumer(Schema.AUTO_CONSUME())
	.topic(autoProducerTopic)
	.subscriptionName("test")
	.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
	.subscribe();
	Message<GenericRecord> message;
	for (int i = 0; i < totalMsgCnt.get(); i++) {
	message = autoConsumer.receive(5, TimeUnit.SECONDS);
	if (message == null) {
	Assert.fail("Failed to receive multiple schema message.");
	}
	log.info("auto consumer get native object class: {}, value: {}",
	message.getValue().getNativeObject().getClass(), message.getValue().getNativeObject());
	checkSchemaForAutoSchema(message);
	}
	}