diff --git a/index.d.ts b/index.d.ts index a24aa75..2d0cef4 100644 --- a/index.d.ts +++ b/index.d.ts @@ -62,6 +62,11 @@ export class Client { subscribe(config: ConsumerConfig): Promise; createReader(config: ReaderConfig): Promise; getPartitionsForTopic(topic: string): Promise; + /** + * 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; close(): Promise; } @@ -246,7 +251,7 @@ export interface EncryptionContext { export interface SchemaInfo { schemaType: SchemaType; name?: string; - schema?: string; + schema?: string | Buffer; properties?: Record; } diff --git a/src/Client.cc b/src/Client.cc index 581b62c..2447d02 100644 --- a/src/Client.cc +++ b/src/Client.cc @@ -22,6 +22,7 @@ #include "Consumer.h" #include "Producer.h" #include "Reader.h" +#include "SchemaInfo.h" #include "ThreadSafeDeferred.h" #include #include @@ -32,6 +33,7 @@ #include #include "pulsar/ClientConfiguration.h" #include +#include #include #include #include @@ -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().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().Utf8Value(); + auto deferred = ThreadSafeDeferred::New(env); + + this->cClient.get()->client->getSchemaInfoAsync( + topic, static_cast(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(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(logMessage->level)); Napi::String file = Napi::String::New(env, logMessage->file); diff --git a/src/Client.h b/src/Client.h index 6ea9a15..312d220 100644 --- a/src/Client.h +++ b/src/Client.h @@ -61,6 +61,7 @@ class Client : public Napi::ObjectWrap { 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); }; diff --git a/src/Client.js b/src/Client.js index 150bbed..7892fd3 100644 --- a/src/Client.js +++ b/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(); } diff --git a/src/SchemaInfo.cc b/src/SchemaInfo.cc index ed09a79..ead4947 100644 --- a/src/SchemaInfo.cc +++ b/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 buffer = schemaInfo.Get(CFG_SCHEMA).As>(); + 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_ptrschemaType, this->name, this->schema, this->properties)); } +std::optional 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::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_writable | napi_enumerable | napi_configurable))); + } + obj.Set(CFG_PROPS, props); + return obj; +} + SchemaInfo::~SchemaInfo() {} diff --git a/src/SchemaInfo.h b/src/SchemaInfo.h index ab746e8..33f6aec 100644 --- a/src/SchemaInfo.h +++ b/src/SchemaInfo.h @@ -23,6 +23,7 @@ #include #include #include +#include #include #include #include @@ -34,6 +35,9 @@ class SchemaInfo { ~SchemaInfo(); void SetProducerSchema(std::shared_ptr cProducerConfiguration); void SetConsumerSchema(std::shared_ptr cConsumerConfiguration); + static std::optional GetSchemaTypeName(pulsar::SchemaType schemaType); + static Napi::Object NewObject(Napi::Env env, const std::string &schemaTypeName, + const pulsar::SchemaInfo &schemaInfo); private: pulsar::SchemaType schemaType; diff --git a/tests/client.test.js b/tests/client.test.js index f871b0d..532a91e 100644 --- a/tests/client.test.js +++ b/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(); + }); + }); }); })(); diff --git a/tstest.ts b/tstest.ts index 6806e2e..c94e49d 100644 --- a/tstest.ts +++ b/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();