diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java index 83f80e2706476..5f4d5f5c0e082 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java @@ -2027,6 +2027,11 @@ remoteAddress, new String(commandGetSchema.getSchemaVersion()), long requestId = commandGetSchema.getRequestId(); SchemaVersion schemaVersion = SchemaVersion.Latest; if (commandGetSchema.hasSchemaVersion()) { + if (commandGetSchema.getSchemaVersion().length == 0) { + commandSender.sendGetSchemaErrorResponse(requestId, ServerError.IncompatibleSchema, + "Empty schema version"); + return; + } schemaVersion = schemaService.versionFromBytes(commandGetSchema.getSchemaVersion()); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java index 3c6f7c73fbc9b..217e76820f15c 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/schema/SchemaTest.java @@ -29,12 +29,13 @@ import static org.testng.Assert.fail; import static org.testng.internal.junit.ArrayAsserts.assertArrayEquals; +import com.google.common.base.Throwables; +import lombok.EqualsAndHashCode; import org.apache.avro.Schema.Parser; - import com.fasterxml.jackson.databind.JsonNode; import com.google.common.collect.Sets; - import java.io.ByteArrayInputStream; +import java.io.Serializable; import java.nio.charset.StandardCharsets; import java.util.Collections; import java.util.HashMap; @@ -44,7 +45,6 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.stream.Collectors; - import lombok.Cleanup; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.client.BKException; @@ -59,7 +59,9 @@ 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.SchemaSerializationException; import org.apache.pulsar.client.api.SubscriptionInitialPosition; +import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.client.api.TypedMessageBuilder; import org.apache.pulsar.client.api.schema.GenericRecord; import org.apache.pulsar.client.api.schema.SchemaDefinition; @@ -1112,4 +1114,72 @@ private void checkSchemaForAutoSchema(Message message) { } } + @Test + public void testAvroSchemaWithHttpLookup() throws Exception { + stopBroker(); + isTcpLookup = false; + setup(); + testEmptySchema(); + } + + @Test + public void testAvroSchemaWithTcpLookup() throws Exception { + stopBroker(); + isTcpLookup = true; + setup(); + testEmptySchema(); + } + + private void testEmptySchema() throws Exception { + final String namespace = "test-namespace-" + randomName(16); + String ns = PUBLIC_TENANT + "/" + namespace; + admin.namespaces().createNamespace(ns, Sets.newHashSet(CLUSTER_NAME)); + + final String autoProducerTopic = getTopicName(ns, "testEmptySchema"); + + @Cleanup + Consumer consumer = pulsarClient + .newConsumer(Schema.AVRO(User.class)) + .topic(autoProducerTopic) + .subscriptionType(SubscriptionType.Shared) + .subscriptionName("sub-1") + .subscribe(); + + @Cleanup + Producer userProducer = pulsarClient + .newProducer(Schema.AVRO(User.class)) + .topic(autoProducerTopic) + .enableBatching(false) + .create(); + + @Cleanup + Producer producer = pulsarClient + .newProducer() + .topic(autoProducerTopic) + .enableBatching(false) + .create(); + + User test = new User("test"); + userProducer.send(test); + producer.send("test".getBytes(StandardCharsets.UTF_8)); + Message message1 = consumer.receive(); + Assert.assertEquals(test, message1.getValue()); + try { + Message message2 = consumer.receive(); + message2.getValue(); + } catch (Throwable ex) { + Assert.assertTrue(Throwables.getRootCause(ex) instanceof SchemaSerializationException); + Assert.assertEquals(Throwables.getRootCause(ex).getMessage(),"Empty schema version"); + } + } + + @EqualsAndHashCode + static class User implements Serializable { + private String name; + public User() {} + public User(String name) { + this.name = name; + } + } + } diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java index 274c83c8c1fc8..190599dfffdbc 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java @@ -32,6 +32,7 @@ import java.util.concurrent.atomic.AtomicLong; import org.apache.commons.lang3.tuple.Pair; import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.api.SchemaSerializationException; import org.apache.pulsar.common.api.proto.CommandGetTopicsOfNamespace.Mode; import org.apache.pulsar.common.api.proto.CommandLookupTopicResponse; import org.apache.pulsar.common.api.proto.CommandLookupTopicResponse.LookupType; @@ -222,9 +223,12 @@ public CompletableFuture> getSchema(TopicName topicName) { @Override public CompletableFuture> getSchema(TopicName topicName, byte[] version) { - InetSocketAddress socketAddress = serviceNameResolver.resolveHost(); CompletableFuture> schemaFuture = new CompletableFuture<>(); - + if (version != null && version.length == 0) { + schemaFuture.completeExceptionally(new SchemaSerializationException("Empty schema version")); + return schemaFuture; + } + InetSocketAddress socketAddress = serviceNameResolver.resolveHost(); client.getCnxPool().getConnection(socketAddress).thenAccept(clientCnx -> { long requestId = client.newRequestId(); ByteBuf request = Commands.newGetSchema(requestId, topicName.toString(), diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpLookupService.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpLookupService.java index e05486cd95dd7..ced0aba9bba10 100644 --- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpLookupService.java +++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpLookupService.java @@ -34,6 +34,7 @@ import org.apache.commons.lang3.tuple.Pair; import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.PulsarClientException.NotFoundException; +import org.apache.pulsar.client.api.SchemaSerializationException; import org.apache.pulsar.client.impl.conf.ClientConfigurationData; import org.apache.pulsar.client.impl.schema.SchemaInfoUtil; import org.apache.pulsar.client.impl.schema.SchemaUtils; @@ -159,6 +160,10 @@ public CompletableFuture> getSchema(TopicName topicName, by String schemaName = topicName.getSchemaName(); String path = String.format("admin/v2/schemas/%s/schema", schemaName); if (version != null) { + if (version.length == 0) { + future.completeExceptionally(new SchemaSerializationException("Empty schema version")); + return future; + } path = String.format("admin/v2/schemas/%s/schema/%s", schemaName, ByteBuffer.wrap(version).getLong());