[feat][client] Expose Client.getSchemaInfo

index.d.ts

@@ -62,6 +62,11 @@ export class Client {
   subscribe(config: ConsumerConfig): Promise<Consumer>;
   createReader(config: ReaderConfig): Promise<Reader>;
   getPartitionsForTopic(topic: string): Promise<string[]>;
+  /**
+   * Fetch the schema registered for `topic` at `version`, such as the writer schema of a
+   * message from `Message.getLongSchemaVersion()`. The schema data is returned as raw bytes.
+   */
+  getSchemaInfo(topic: string, version: number): Promise<SchemaInfo & { schema: Buffer }>;
   close(): Promise<null>;
 }
 
@@ -246,7 +251,7 @@ export interface EncryptionContext {
 export interface SchemaInfo {
   schemaType: SchemaType;
   name?: string;
-  schema?: string;
+  schema?: string | Buffer;
   properties?: Record<string, string>;
 }
 

src/Client.cc

@@ -22,6 +22,7 @@
 #include "Consumer.h"
 #include "Producer.h"
 #include "Reader.h"
+#include "SchemaInfo.h"
 #include "ThreadSafeDeferred.h"
 #include <pulsar/AutoClusterFailover.h>
 #include <pulsar/Client.h>
@@ -32,6 +33,7 @@
 #include <pulsar/c/result.h>
 #include "pulsar/ClientConfiguration.h"
 #include <chrono>
+#include <cmath>
 #include <cstdint>
 #include <memory>
 #include <optional>
@@ -293,7 +295,7 @@ Napi::Object Client::Init(Napi::Env env, Napi::Object exports) {
        InstanceMethod("createProducer", &Client::CreateProducer),
        InstanceMethod("subscribe", &Client::Subscribe), InstanceMethod("createReader", &Client::CreateReader),
        InstanceMethod("getPartitionsForTopic", &Client::GetPartitionsForTopic),
-       InstanceMethod("close", &Client::Close)});
+       InstanceMethod("getSchemaInfo", &Client::GetSchemaInfo), InstanceMethod("close", &Client::Close)});
 
   constructor = Napi::Persistent(func);
   constructor.SuppressDestruct();
@@ -523,6 +525,43 @@ Napi::Value Client::GetPartitionsForTopic(const Napi::CallbackInfo &info) {
   return deferred->Promise();
 }
 
+Napi::Value Client::GetSchemaInfo(const Napi::CallbackInfo &info) {
+  Napi::Env env = info.Env();
+  constexpr double maxSafeInteger = 9007199254740991;  // Number.MAX_SAFE_INTEGER
+  // Converting a fractional, negative, or non-finite number would silently fetch a different version.
+  bool validVersion = info.Length() >= 2 && info[1].IsNumber();
+  double version = validVersion ? info[1].As<Napi::Number>().DoubleValue() : 0;
+  validVersion = validVersion && version >= 0 && version <= maxSafeInteger && std::trunc(version) == version;
+  if (!info[0].IsString() || !validVersion) {
+    Napi::TypeError::New(env,
+                         "getSchemaInfo expects a topic (string) and a schema version (non-negative integer)")
+        .ThrowAsJavaScriptException();
+    return env.Null();
+  }
+  std::string topic = info[0].As<Napi::String>().Utf8Value();
+  auto deferred = ThreadSafeDeferred::New(env);
+
+  this->cClient.get()->client->getSchemaInfoAsync(
+      topic, static_cast<int64_t>(version),
+      [deferred](pulsar::Result result, const pulsar::SchemaInfo &schemaInfo) {
+        if (result != pulsar::ResultOk) {
+          deferred->Reject(std::string("Failed to get schema info: ") + pulsar::strResult(result));
+          return;
+        }
+        auto schemaTypeName = SchemaInfo::GetSchemaTypeName(schemaInfo.getSchemaType());
+        if (!schemaTypeName) {
+          deferred->Reject("Failed to get schema info: unsupported schema type " +
+                           std::to_string(static_cast<int>(schemaInfo.getSchemaType())));
+          return;
+        }
+        deferred->Resolve([schemaInfo, schemaTypeName](const Napi::Env env) {
+          return SchemaInfo::NewObject(env, *schemaTypeName, schemaInfo);
+        });
+      });
+
+  return deferred->Promise();
+}
+
 void LogMessageProxy(Napi::Env env, Napi::Function jsCallback, struct LogMessage *logMessage) {
   Napi::Number logLevel = Napi::Number::New(env, static_cast<double>(logMessage->level));
   Napi::String file = Napi::String::New(env, logMessage->file);

src/Client.h

@@ -61,6 +61,7 @@ class Client : public Napi::ObjectWrap<Client> {
   Napi::Value Subscribe(const Napi::CallbackInfo &info);
   Napi::Value CreateReader(const Napi::CallbackInfo &info);
   Napi::Value GetPartitionsForTopic(const Napi::CallbackInfo &info);
+  Napi::Value GetSchemaInfo(const Napi::CallbackInfo &info);
   Napi::Value Close(const Napi::CallbackInfo &info);
 };
 

src/Client.js

@@ -48,6 +48,10 @@ class Client {
     return this.client.getPartitionsForTopic(params);
   }
 
+  getSchemaInfo(topic, version) {
+    return this.client.getSchemaInfo(topic, version);
+  }
+
   close() {
     return this.client.close();
   }

src/SchemaInfo.cc

@@ -70,6 +70,9 @@ SchemaInfo::SchemaInfo(const Napi::Object &schemaInfo)
   }
   if (schemaInfo.Has(CFG_SCHEMA) && schemaInfo.Get(CFG_SCHEMA).IsString()) {
     this->schema = schemaInfo.Get(CFG_SCHEMA).ToString().Utf8Value();
+  } else if (schemaInfo.Has(CFG_SCHEMA) && schemaInfo.Get(CFG_SCHEMA).IsBuffer()) {
+    Napi::Buffer<char> buffer = schemaInfo.Get(CFG_SCHEMA).As<Napi::Buffer<char>>();
+    this->schema = std::string(buffer.Data(), buffer.Length());
   }
   if (schemaInfo.Has(CFG_PROPS) && schemaInfo.Get(CFG_PROPS).IsObject()) {
     Napi::Object propObj = schemaInfo.Get(CFG_PROPS).ToObject();
@@ -93,4 +96,32 @@ void SchemaInfo::SetConsumerSchema(std::shared_ptr<pulsar_consumer_configuration
       pulsar::SchemaInfo(this->schemaType, this->name, this->schema, this->properties));
 }
 
+std::optional<std::string> SchemaInfo::GetSchemaTypeName(pulsar::SchemaType schemaType) {
+  for (const auto &entry : SCHEMA_TYPE) {
+    if (entry.second == schemaType) {
+      return entry.first;
+    }
+  }
+  return std::nullopt;
+}
+
+Napi::Object SchemaInfo::NewObject(Napi::Env env, const std::string &schemaTypeName,
+                                   const pulsar::SchemaInfo &schemaInfo) {
+  Napi::Object obj = Napi::Object::New(env);
+  obj.Set(CFG_SCHEMA_TYPE, Napi::String::New(env, schemaTypeName));
+  obj.Set(CFG_NAME, Napi::String::New(env, schemaInfo.getName()));
+  // Schema data is bytes: KeyValue schemas, for one, are binary and would not survive a UTF-8 string.
+  const std::string &schema = schemaInfo.getSchema();
+  obj.Set(CFG_SCHEMA, Napi::Buffer<char>::Copy(env, schema.data(), schema.size()));
+  Napi::Object props = Napi::Object::New(env);
+  for (const auto &prop : schemaInfo.getProperties()) {
+    // Define own properties: Set would route a "__proto__" key to the prototype setter and drop it.
+    props.DefineProperty(Napi::PropertyDescriptor::Value(
+        prop.first, Napi::String::New(env, prop.second),
+        static_cast<napi_property_attributes>(napi_writable | napi_enumerable | napi_configurable)));
+  }
+  obj.Set(CFG_PROPS, props);
+  return obj;
+}
+
 SchemaInfo::~SchemaInfo() {}

src/SchemaInfo.h

@@ -23,6 +23,7 @@
 #include <map>
 #include <memory>
 #include <napi.h>
+#include <optional>
 #include <pulsar/Schema.h>
 #include <pulsar/c/consumer_configuration.h>
 #include <pulsar/c/producer_configuration.h>
@@ -34,6 +35,9 @@ class SchemaInfo {
   ~SchemaInfo();
   void SetProducerSchema(std::shared_ptr<pulsar_producer_configuration_t> cProducerConfiguration);
   void SetConsumerSchema(std::shared_ptr<pulsar_consumer_configuration_t> cConsumerConfiguration);
+  static std::optional<std::string> GetSchemaTypeName(pulsar::SchemaType schemaType);
+  static Napi::Object NewObject(Napi::Env env, const std::string &schemaTypeName,
+                                const pulsar::SchemaInfo &schemaInfo);
 
  private:
   pulsar::SchemaType schemaType;

tests/client.test.js

@@ -131,5 +131,164 @@ const baseUrl = 'http://localhost:8080';
         await client.close();
       });
     });
+
+    describe('test getSchemaInfo', () => {
+      test('Client/getSchemaInfo', async () => {
+        const client = new Pulsar.Client({
+          serviceUrl: 'pulsar://localhost:6650',
+          operationTimeoutSeconds: 30,
+        });
+
+        const definitionV1 = {
+          type: 'record',
+          name: 'User',
+          fields: [{ name: 'name', type: 'string' }],
+        };
+        // Compatible evolution of definitionV1: adds a field with a default.
+        const definitionV2 = {
+          type: 'record',
+          name: 'User',
+          fields: [
+            { name: 'name', type: 'string' },
+            { name: 'age', type: 'int', default: 0 },
+          ],
+        };
+
+        const topic = `persistent://public/default/get-schema-info-${Date.now()}`;
+        const consumer = await client.subscribe({
+          topic,
+          subscription: 'sub1',
+          schema: { schemaType: 'Avro', schema: JSON.stringify(definitionV1) },
+        });
+        const producerV1 = await client.createProducer({
+          topic,
+          schema: { schemaType: 'Avro', schema: JSON.stringify(definitionV1) },
+        });
+        const producerV2 = await client.createProducer({
+          topic,
+          schema: { schemaType: 'Avro', schema: JSON.stringify(definitionV2) },
+        });
+        // The payloads are not decoded here, only the schema version each message carries.
+        await producerV1.send({ data: Buffer.from('v1') });
+        await producerV2.send({ data: Buffer.from('v2') });
+        const msgV1 = await consumer.receive();
+        const msgV2 = await consumer.receive();
+        consumer.acknowledge(msgV1);
+        consumer.acknowledge(msgV2);
+
+        const schemaV1 = await client.getSchemaInfo(topic, msgV1.getLongSchemaVersion());
+        const schemaV2 = await client.getSchemaInfo(topic, msgV2.getLongSchemaVersion());
+        expect(schemaV1.schemaType).toBe('Avro');
+        expect(JSON.parse(schemaV1.schema.toString())).toEqual(definitionV1);
+        expect(schemaV2.schemaType).toBe('Avro');
+        expect(JSON.parse(schemaV2.schema.toString())).toEqual(definitionV2);
+
+        await expect(client.getSchemaInfo(topic, 99))
+          .rejects.toThrow('Failed to get schema info: ');
+
+        await producerV1.close();
+        await producerV2.close();
+        await consumer.close();
+        await client.close();
+      });
+
+      test('KeyValue schema', async () => {
+        const client = new Pulsar.Client({
+          serviceUrl: 'pulsar://localhost:6650',
+          operationTimeoutSeconds: 30,
+        });
+
+        // A key schema longer than 127 bytes, so its length prefix is not valid UTF-8.
+        const keyDefinition = {
+          type: 'record',
+          name: 'Key',
+          fields: [{ name: 'id', type: 'string', doc: 'x'.repeat(150) }],
+        };
+        const valueDefinition = {
+          type: 'record',
+          name: 'Value',
+          fields: [{ name: 'count', type: 'int' }],
+        };
+        const topicName = `get-schema-info-key-value-${Date.now()}`;
+        const topic = `persistent://public/default/${topicName}`;
+        // The admin API takes the key and value schemas as JSON and stores them in the binary
+        // KeyValue layout: [key length][key schema][value length][value schema].
+        const postSchemaRes = await httpUtils.request(`${baseUrl}/admin/v2/schemas/public/default/${topicName}/schema`, {
+          headers: { 'Content-Type': 'application/json' },
+          data: {
+            type: 'KEY_VALUE',
+            schema: JSON.stringify({ key: keyDefinition, value: valueDefinition }),
+            properties: {
+              'key.schema.type': 'AVRO',
+              'key.schema.name': 'Key',
+              'key.schema.properties': '{}',
+              'value.schema.type': 'AVRO',
+              'value.schema.name': 'Value',
+              'value.schema.properties': '{}',
+              'kv.encoding.type': 'SEPARATED',
+            },
+          },
+          method: 'POST',
+        });
+        expect(postSchemaRes.statusCode).toBe(200);
+
+        const schema = await client.getSchemaInfo(topic, 0);
+        expect(schema.schemaType).toBe('KeyValue');
+        const data = schema.schema;
+        const keyLength = data.readInt32BE(0);
+        const valueLength = data.readInt32BE(4 + keyLength);
+        expect(keyLength).toBeGreaterThan(127);
+        expect(data.length).toBe(8 + keyLength + valueLength);
+        expect(JSON.parse(data.subarray(4, 4 + keyLength).toString())).toEqual(keyDefinition);
+        expect(JSON.parse(data.subarray(8 + keyLength).toString())).toEqual(valueDefinition);
+
+        // The fetched schema is accepted back as-is: the producer reuses version 0.
+        const producer = await client.createProducer({ topic, schema });
+        await expect(client.getSchemaInfo(topic, 1)).rejects.toThrow('Failed to get schema info: ');
+
+        await producer.close();
+        await client.close();
+      });
+
+      test('Schema properties', async () => {
+        const client = new Pulsar.Client({
+          serviceUrl: 'pulsar://localhost:6650',
+          operationTimeoutSeconds: 30,
+        });
+
+        const topic = `persistent://public/default/get-schema-info-properties-${Date.now()}`;
+        // JSON.parse creates an own "__proto__" key, which a plain assignment would not.
+        const properties = JSON.parse('{"owner":"team-a","__proto__":"value"}');
+        const producer = await client.createProducer({
+          topic,
+          schema: { schemaType: 'Json', schema: JSON.stringify({ type: 'string' }), properties },
+        });
+
+        const schema = await client.getSchemaInfo(topic, 0);
+        expect(Object.entries(schema.properties).sort()).toEqual([
+          ['__proto__', 'value'],
+          ['owner', 'team-a'],
+        ]);
+
+        await producer.close();
+        await client.close();
+      });
+
+      test('Invalid arguments', async () => {
+        const client = new Pulsar.Client({
+          serviceUrl: 'pulsar://localhost:6650',
+          operationTimeoutSeconds: 30,
+        });
+
+        const topic = 'persistent://public/default/my-topic';
+        const expectedError = 'getSchemaInfo expects a topic (string) and a schema version (non-negative integer)';
+        expect(() => client.getSchemaInfo(topic)).toThrow(expectedError);
+        [0.5, -1, NaN, Infinity, 2 ** 53, '0'].forEach((version) => {
+          expect(() => client.getSchemaInfo(topic, version)).toThrow(expectedError);
+        });
+
+        await client.close();
+      });
+    });
   });
 })();

tstest.ts

@@ -310,6 +310,8 @@ import Pulsar = require('./index');
   const isReplicated: boolean = message1.isReplicated();
   const replicatedFrom: string = message1.getReplicatedFrom();
   const schemaVersion: number | null = message1.getLongSchemaVersion();
+  const schemaInfo: Pulsar.SchemaInfo = await client.getSchemaInfo('persistent://public/default/my-topic', 0);
+  const schemaData: Buffer = (await client.getSchemaInfo('persistent://public/default/my-topic', 0)).schema;
   const partitionKey: string = message1.getPartitionKey();
 
   const message3: Pulsar.Message = await reader1.readNext();