|
18 | 18 | */ |
19 | 19 | package org.apache.pulsar.client.api; |
20 | 20 |
|
| 21 | +import lombok.AllArgsConstructor; |
| 22 | +import lombok.Builder; |
| 23 | +import lombok.Data; |
| 24 | +import lombok.NoArgsConstructor; |
| 25 | + |
21 | 26 | import static java.nio.charset.StandardCharsets.UTF_8; |
22 | 27 | import static org.testng.Assert.assertEquals; |
23 | 28 | import static org.testng.Assert.assertNotNull; |
24 | 29 | import static org.testng.Assert.assertTrue; |
25 | 30 |
|
| 31 | +import org.apache.avro.reflect.ReflectData; |
| 32 | +import org.apache.avro.Schema.Parser; |
| 33 | +import org.apache.pulsar.broker.service.schema.LongSchemaVersion; |
26 | 34 | import org.apache.pulsar.client.api.PulsarClientException.IncompatibleSchemaException; |
| 35 | +import org.apache.pulsar.client.api.PulsarClientException.InvalidMessageException; |
27 | 36 | import org.apache.pulsar.client.api.schema.GenericRecord; |
| 37 | +import org.apache.pulsar.client.impl.ProducerBase; |
| 38 | +import org.apache.pulsar.client.impl.schema.writer.AvroWriter; |
| 39 | +import org.apache.pulsar.common.protocol.schema.SchemaVersion; |
28 | 40 | import org.apache.pulsar.common.schema.KeyValue; |
29 | 41 | import org.apache.pulsar.common.schema.KeyValueEncodingType; |
| 42 | +import org.apache.pulsar.common.schema.SchemaInfo; |
30 | 43 | import org.testng.Assert; |
31 | 44 | import org.testng.annotations.AfterMethod; |
32 | 45 | import org.testng.annotations.BeforeMethod; |
33 | 46 | import org.testng.annotations.DataProvider; |
34 | 47 | import org.testng.annotations.Factory; |
35 | 48 | import org.testng.annotations.Test; |
36 | 49 |
|
| 50 | +import java.io.ByteArrayInputStream; |
| 51 | +import java.util.Arrays; |
| 52 | +import java.util.List; |
| 53 | +import java.util.concurrent.TimeUnit; |
| 54 | + |
37 | 55 | public class SimpleSchemaTest extends ProducerConsumerBase { |
38 | 56 |
|
39 | 57 | @DataProvider(name = "batchingModes") |
@@ -94,26 +112,30 @@ public void testString() throws Exception { |
94 | 112 | } |
95 | 113 | } |
96 | 114 |
|
| 115 | + @Data |
| 116 | + @Builder |
| 117 | + @NoArgsConstructor |
| 118 | + @AllArgsConstructor |
97 | 119 | static class V1Data { |
98 | 120 | int i; |
| 121 | + } |
99 | 122 |
|
100 | | - V1Data() { |
101 | | - this.i = 0; |
102 | | - } |
103 | | - |
104 | | - V1Data(int i) { |
105 | | - this.i = i; |
106 | | - } |
107 | | - |
108 | | - @Override |
109 | | - public int hashCode() { |
110 | | - return i; |
111 | | - } |
| 123 | + @Data |
| 124 | + @Builder |
| 125 | + @NoArgsConstructor |
| 126 | + @AllArgsConstructor |
| 127 | + static class V2Data { |
| 128 | + int i; |
| 129 | + Integer j; |
| 130 | + } |
112 | 131 |
|
113 | | - @Override |
114 | | - public boolean equals(Object other) { |
115 | | - return (other instanceof V1Data) && i == ((V1Data)other).i; |
116 | | - } |
| 132 | + @Data |
| 133 | + @Builder |
| 134 | + @NoArgsConstructor |
| 135 | + @AllArgsConstructor |
| 136 | + static class IncompatibleData { |
| 137 | + int i; |
| 138 | + int j; |
117 | 139 | } |
118 | 140 |
|
119 | 141 | @Test |
@@ -185,6 +207,172 @@ public void newProducerWithoutSchemaOnTopicWithSchema() throws Exception { |
185 | 207 | } |
186 | 208 | } |
187 | 209 |
|
| 210 | + @Test |
| 211 | + public void newProducerForMessageSchemaOnTopicWithMultiVersionSchema() throws Exception { |
| 212 | + String topic = "my-property/my-ns/schema-test"; |
| 213 | + Schema<V1Data> v1Schema = Schema.AVRO(V1Data.class); |
| 214 | + byte[] v1SchemaBytes = v1Schema.getSchemaInfo().getSchema(); |
| 215 | + AvroWriter<V1Data> v1Writer = new AvroWriter<>( |
| 216 | + new Parser().parse(new ByteArrayInputStream(v1SchemaBytes))); |
| 217 | + Schema<V2Data> v2Schema = Schema.AVRO(V2Data.class); |
| 218 | + byte[] v2SchemaBytes = v2Schema.getSchemaInfo().getSchema(); |
| 219 | + AvroWriter<V2Data> v2Writer = new AvroWriter<>( |
| 220 | + new Parser().parse(new ByteArrayInputStream(v2SchemaBytes))); |
| 221 | + try (Producer<V1Data> ignored = pulsarClient.newProducer(v1Schema) |
| 222 | + .topic(topic).create()) { |
| 223 | + } |
| 224 | + try (Producer<V2Data> p = pulsarClient.newProducer(Schema.AVRO(V2Data.class)) |
| 225 | + .topic(topic).create()) { |
| 226 | + p.send(new V2Data(-1, -1)); |
| 227 | + } |
| 228 | + V1Data dataV1 = new V1Data(2); |
| 229 | + V2Data dataV2 = new V2Data(3, 5); |
| 230 | + byte[] contentV1 = v1Writer.write(dataV1); |
| 231 | + byte[] contentV2 = v2Writer.write(dataV2); |
| 232 | + try (Producer<byte[]> p = pulsarClient.newProducer(Schema.AUTO_PRODUCE_BYTES()) |
| 233 | + .topic(topic).create(); |
| 234 | + Consumer<V2Data> c = pulsarClient.newConsumer(v2Schema) |
| 235 | + .topic(topic) |
| 236 | + .subscriptionName("sub1").subscribe()) { |
| 237 | + Assert.expectThrows(SchemaSerializationException.class, () -> p.send(contentV1)); |
| 238 | + |
| 239 | + ((ProducerBase<byte[]>)p).newMessage(Schema.AUTO_PRODUCE_BYTES(Schema.AVRO(V1Data.class))) |
| 240 | + .value(contentV1).send(); |
| 241 | + p.send(contentV2); |
| 242 | + Message<V2Data> msg1 = c.receive(); |
| 243 | + V2Data msg1Value = msg1.getValue(); |
| 244 | + Assert.assertEquals(dataV1.i, msg1Value.i); |
| 245 | + Assert.assertNull(msg1Value.j); |
| 246 | + Assert.assertEquals(msg1.getSchemaVersion(), new LongSchemaVersion(0).bytes()); |
| 247 | + |
| 248 | + Message<V2Data> msg2 = c.receive(); |
| 249 | + Assert.assertEquals(dataV2, msg2.getValue()); |
| 250 | + Assert.assertEquals(msg2.getSchemaVersion(), new LongSchemaVersion(1).bytes()); |
| 251 | + |
| 252 | + try { |
| 253 | + ((ProducerBase<byte[]>)p).newMessage(Schema.BYTES).value(contentV1).send(); |
| 254 | + if (schemaValidationEnforced) { |
| 255 | + Assert.fail("Shouldn't be able to send to a schema'd topic with no schema" |
| 256 | + + " if SchemaValidationEnabled is enabled"); |
| 257 | + } |
| 258 | + Message<V2Data> msg3 = c.receive(); |
| 259 | + Assert.assertEquals(msg3.getSchemaVersion(), SchemaVersion.Empty.bytes()); |
| 260 | + } catch (PulsarClientException e) { |
| 261 | + if (schemaValidationEnforced) { |
| 262 | + Assert.assertTrue(e instanceof IncompatibleSchemaException); |
| 263 | + } else { |
| 264 | + Assert.fail("Shouldn't throw IncompatibleSchemaException" |
| 265 | + + " if SchemaValidationEnforced is disabled"); |
| 266 | + } |
| 267 | + } |
| 268 | + } |
| 269 | + } |
| 270 | + |
| 271 | + @Test |
| 272 | + public void newProducerForMessageSchemaOnTopicInitialWithNoSchema() throws Exception { |
| 273 | + String topic = "my-property/my-ns/schema-test"; |
| 274 | + Schema<V1Data> v1Schema = Schema.AVRO(V1Data.class); |
| 275 | + byte[] v1SchemaBytes = v1Schema.getSchemaInfo().getSchema(); |
| 276 | + AvroWriter<V1Data> v1Writer = new AvroWriter<>( |
| 277 | + new Parser().parse(new ByteArrayInputStream(v1SchemaBytes))); |
| 278 | + Schema<V2Data> v2Schema = Schema.AVRO(V2Data.class); |
| 279 | + byte[] v2SchemaBytes = v2Schema.getSchemaInfo().getSchema(); |
| 280 | + AvroWriter<V2Data> v2Writer = new AvroWriter<>( |
| 281 | + new Parser().parse(new ByteArrayInputStream(v2SchemaBytes))); |
| 282 | + try (Producer<byte[]> p = pulsarClient.newProducer() |
| 283 | + .topic(topic).create(); |
| 284 | + Consumer<byte[]> c = pulsarClient.newConsumer() |
| 285 | + .topic(topic) |
| 286 | + .subscriptionName("sub1").subscribe()) { |
| 287 | + for (int i = 0; i < 2; ++i) { |
| 288 | + V1Data dataV1 = new V1Data(i); |
| 289 | + V2Data dataV2 = new V2Data(i, -i); |
| 290 | + byte[] contentV1 = v1Writer.write(dataV1); |
| 291 | + byte[] contentV2 = v2Writer.write(dataV2); |
| 292 | + ((ProducerBase<byte[]>) p).newMessage(Schema.AUTO_PRODUCE_BYTES(v1Schema)) |
| 293 | + .value(contentV1).send(); |
| 294 | + Message<byte[]> msg1 = c.receive(); |
| 295 | + Assert.assertEquals(msg1.getSchemaVersion(), new LongSchemaVersion(0).bytes()); |
| 296 | + Assert.assertEquals(msg1.getData(), contentV1); |
| 297 | + ((ProducerBase<byte[]>) p).newMessage(Schema.AUTO_PRODUCE_BYTES(v2Schema)) |
| 298 | + .value(contentV2).send(); |
| 299 | + Message<byte[]> msg2 = c.receive(); |
| 300 | + Assert.assertEquals(msg2.getSchemaVersion(), new LongSchemaVersion(1).bytes()); |
| 301 | + Assert.assertEquals(msg2.getData(), contentV2); |
| 302 | + } |
| 303 | + } |
| 304 | + |
| 305 | + List<SchemaInfo> allSchemas = admin.schemas().getAllSchemas(topic); |
| 306 | + Assert.assertEquals(allSchemas, Arrays.asList(v1Schema.getSchemaInfo(), |
| 307 | + v2Schema.getSchemaInfo())); |
| 308 | + } |
| 309 | + |
| 310 | + @Test |
| 311 | + public void newProducerForMessageSchemaWithBatch() throws Exception { |
| 312 | + String topic = "my-property/my-ns/schema-test"; |
| 313 | + Consumer<V2Data> c = pulsarClient.newConsumer(Schema.AVRO(V2Data.class)) |
| 314 | + .topic(topic) |
| 315 | + .subscriptionName("sub1").subscribe(); |
| 316 | + Producer<byte[]> p = pulsarClient.newProducer(Schema.AUTO_PRODUCE_BYTES()) |
| 317 | + .topic(topic) |
| 318 | + .enableBatching(true) |
| 319 | + .batchingMaxPublishDelay(10, TimeUnit.SECONDS).create(); |
| 320 | + AvroWriter<V1Data> v1DataAvroWriter = new AvroWriter<>( |
| 321 | + ReflectData.AllowNull.get().getSchema(V1Data.class)); |
| 322 | + AvroWriter<V2Data> v2DataAvroWriter = new AvroWriter<>( |
| 323 | + ReflectData.AllowNull.get().getSchema(V2Data.class)); |
| 324 | + AvroWriter<IncompatibleData> incompatibleDataAvroWriter = new AvroWriter<>( |
| 325 | + ReflectData.AllowNull.get().getSchema(IncompatibleData.class)); |
| 326 | + int total = 20; |
| 327 | + int batch = 5; |
| 328 | + int incompatible = 3; |
| 329 | + for (int i = 0; i < total; ++i) { |
| 330 | + if (i / batch % 2 == 0) { |
| 331 | + byte[] content = v1DataAvroWriter.write(new V1Data(i)); |
| 332 | + ((ProducerBase<byte[]>)p).newMessage(Schema.AUTO_PRODUCE_BYTES(Schema.AVRO(V1Data.class))) |
| 333 | + .value(content).sendAsync(); |
| 334 | + } else { |
| 335 | + byte[] content = v2DataAvroWriter.write(new V2Data(i, i + total)); |
| 336 | + ((ProducerBase<byte[]>)p).newMessage(Schema.AUTO_PRODUCE_BYTES(Schema.AVRO(V2Data.class))) |
| 337 | + .value(content).sendAsync(); |
| 338 | + } |
| 339 | + if ((i + 1) % incompatible == 0) { |
| 340 | + byte[] content = incompatibleDataAvroWriter.write(new IncompatibleData(-i, -i)); |
| 341 | + try { |
| 342 | + ((ProducerBase<byte[]>)p).newMessage(Schema.AUTO_PRODUCE_BYTES(Schema.AVRO(IncompatibleData.class))) |
| 343 | + .value(content).send(); |
| 344 | + } catch (Exception e) { |
| 345 | + Assert.assertTrue(e instanceof IncompatibleSchemaException, e.getMessage()); |
| 346 | + } |
| 347 | + } |
| 348 | + } |
| 349 | + p.flush(); |
| 350 | + for (int i = 0; i < total; ++i) { |
| 351 | + V2Data value = c.receive().getValue(); |
| 352 | + if (i / batch % 2 == 0) { |
| 353 | + Assert.assertNull(value.j); |
| 354 | + Assert.assertEquals(value.i, i); |
| 355 | + } else { |
| 356 | + Assert.assertEquals(value, new V2Data(i, i + total)); |
| 357 | + } |
| 358 | + } |
| 359 | + c.close(); |
| 360 | + } |
| 361 | + |
| 362 | + @Test |
| 363 | + public void newProducerWithMultipleSchemaDisabled() throws Exception { |
| 364 | + String topic = "my-property/my-ns/schema-test"; |
| 365 | + AvroWriter<V1Data> v1DataAvroWriter = new AvroWriter<>( |
| 366 | + ReflectData.AllowNull.get().getSchema(V1Data.class)); |
| 367 | + try (Producer<byte[]> p = pulsarClient.newProducer() |
| 368 | + .topic(topic) |
| 369 | + .enableMultiSchema(false).create()) { |
| 370 | + Assert.assertThrows(InvalidMessageException.class, |
| 371 | + () -> ((ProducerBase<byte[]>)p).newMessage(Schema.AUTO_PRODUCE_BYTES(Schema.AVRO(V1Data.class))) |
| 372 | + .value(v1DataAvroWriter.write(new V1Data(0))).send()); |
| 373 | + } |
| 374 | + } |
| 375 | + |
188 | 376 | @Test |
189 | 377 | public void newConsumerWithSchemaOnNewTopic() throws Exception { |
190 | 378 | String topic = "my-property/my-ns/schema-test"; |
|
0 commit comments