add support for JSON-schema and Protobuf (solves #79) #93

Malkiz · 2021-02-16T15:21:31Z

solves #79

Changes in this PR:

refactor the Avro specific logic into a dedicated class
add logic for supporting JSON-schema and Protobuf

… code to make way for other schema types/de(serialization) formats such as proto and json schema

malkiz-rapid · 2021-02-18T08:31:10Z

src/SchemaRegistry.ts

  }

-  public async decode(buffer: Buffer): Promise<any> {
+  public async decode(buffer: Buffer, serdesOpts?: {}): Promise<any> {


MAYBE: make the type of serdesOpts be an OR type of the different serdes types

MAYBE: pass in a getter function schema => options

Another option would be to have it be:

interface SerdesOptions { [SchemaType.AVRO]?: AvroSerdesOptions [SchemaType.JSON]?: JsonSerdesOptions [SchemaType.PROTO]?: ProtoSerdesOptions }

That way the interface remains pretty simple while still allowing you to pass in options for both Protobuf and Avro, in case you are consuming both.

Having a union of all types is problematic because it means that you can only pass in options for a single schema type. Using an intersection type is problematic because the different schema types can have the same option names even though they mean different things or need different values.

malkiz-rapid · 2021-02-18T08:42:15Z

src/AvroSerdes.ts

+    return subject
+  }
+
+  public serialize(schema: ConfluentSchema, payload: any, opts: any): Buffer {


actual AVRO type for opts

malkiz-rapid · 2021-02-18T08:47:43Z

src/JsonSerdes.ts

+  }
+
+  // @ts-ignore
+  public getSubject(schema: ConfluentSchema, separator: string): ConfluentSubject {


NOTE: add implementation according to the Java implementation of the TopicNameStrategy (the default)

https://github.com/confluentinc/schema-registry/blob/master/schema-serializer/src/main/java/io/confluent/kafka/serializers/subject/TopicNameStrategy.java#L39

malkiz-rapid · 2021-02-18T08:50:05Z

src/JsonSerdes.ts

+export default class JsonSerdes implements Serdes {
+  private getJsonSchema(schema: ConfluentSchema, opts?: any) {
+    const ajv = new Ajv(opts)
+    const validate = ajv.compile(JSON.parse(schema.schemaString))


maybe we should cache the validate function somewhere, to save the compilation time

See the "Code Generation Performance" header for relevant context: https://openjsf.org/blog/2021/02/11/project-news-ajv-version-7-big-changes-and-improvements/

malkiz-rapid · 2021-02-18T08:50:29Z

src/JsonSerdes.ts

+  private validatePayload(schema: ConfluentSchema, payload: any, opts: any) {
+    const validate = this.getJsonSchema(schema, opts)
+    if (!validate(payload)) {
+      throw Error(validate.errors as any)


throw a specific error (validation error)

malkiz-rapid · 2021-02-18T09:44:57Z

src/SchemaRegistry.ts

-      if (!schema.namespace) {
-        throw new ConfluentSchemaRegistryArgumentError(`Invalid namespace: ${schema.namespace}`)
-      }
+      throw new ConfluentSchemaRegistryArgumentError(


remove this catch block, and throw specific messages from inside the serdes implementations

malkiz-rapid · 2021-02-18T09:52:06Z

src/SchemaRegistry.ts

    this.cache.setSchema(registeredSchema.id, schema)

    return registeredSchema
  }

-  public async getSchema(registryId: number): Promise<Schema> {
+  public async getSchema(registryId: number): Promise<ConfluentSchema> {


add a private function to return ConfluentSchema.
revert the return type of getSchema back to Schema which will be a union type of the different implementations

malkiz-rapid · 2021-02-18T09:55:52Z

src/@types.ts

  name: string
  namespace?: string
  type: 'record'
  fields: any[]
 }
-export interface Schema extends RawSchema {
+
+export interface AvroSchema extends RawAvroSchema {


implement these 3 methods for the 2 other types

malkiz-rapid · 2021-02-18T09:58:00Z

src/SchemaRegistry.ts

@@ -146,11 +163,14 @@ export default class SchemaRegistry {
    return id
  }

-  public async getRegistryIdBySchema(subject: string, schema: Schema): Promise<number> {
+  public async getRegistryIdBySchema(subject: string, schema: ConfluentSchema): Promise<number> {


do the same thing with the union type to allow backwards compatibility

malkiz-rapid · 2021-02-18T10:01:29Z

src/encoder.ts

-const DEFAULT_OFFSET = 0
-
-// Based on https://github.com/mtth/avsc/issues/140
-const collectInvalidPaths = (schema: Schema, jsonPayload: object) => {


migrate this logic to AvroSerdes

package.json

Nevon · 2021-02-24T12:01:59Z

src/@types.ts

+}
+
+export type AvroOptions = Partial<ForSchemaOptions>
+export type JsonOptions = any // FIXME:


This should be ConstructorParameters<typeof Ajv>[0] (which I think is InstanceOptions from Ajv)

Nevon · 2021-02-24T12:11:34Z

src/@types.ts

+
+export type AvroOptions = Partial<ForSchemaOptions>
+export type JsonOptions = any // FIXME:
+export type ProtoOptions = any // FIXME:


This one is a little trickier. Currently we are expecting an object like { messageName: string }, which doesn't map to any options in Protobufjs itself.

Given that Protobufjs doesn't have any constructor options or anything that we could use here, I suggest that we make this just { messageName: string }, and then in the future if people have a need to pass IParseOptions or something we can extend the type like { messageName: string, parse: IParseOptions } etc.

src/@types.ts

src/AvroHelper.ts

Nevon · 2021-02-24T12:16:55Z

src/JsonHelper.ts

+    schema: Schema,
+    separator: string,
+  ): ConfluentSubject {
+    throw Error('not implemented yet')


Did you have a chance to look into what the Java client does for this and Protobuf?

If not, and we leave them unimplemented for now, let's at least throw a ConfluentSchemaRegistryError instead of a generic error, unless there's something higher up that catches it and re-throws.

no I didn't check how the Java implementation does it.
I'll change it to ConfluentSchemaRegistryError

Nevon · 2021-02-24T15:21:23Z

Awesome job! There'll be a bit of follow-up work before we can release, such as updating the documentation, but I'll create a new issue for that.

Dan Katz and others added 30 commits September 7, 2020 15:01

separating avro specifics from encoder/decoder cache and main library…

927b66a

… code to make way for other schema types/de(serialization) formats such as proto and json schema

removing bc unneeded

97e6f45

minor cleanup of schema type

dd9a09e

install protobufjs and add protoserdes

800529d

use a dynamic serdes according to the current schemaType

1806dd4

test

5f2d986

5.5.3

da5b7f9

JSON

85df64e

protobuf

3c83a30

Merge branch 'master' into serdes-refactor

18e5f75

proto impl

9beccef

Merge branch 'serdes-refactor' into proto

85fdbf6

Merge remote-tracking branch 'origin/proto' into serdes-refactor

8221b6c

tests

5a2ede2

tests

b1d4922

proto

56b4e29

proto

8840dac

lint

c346d9d

tests

fa5b130

proto

d1fd1ad

test

93fd688

proto

4403eab

json

1f96069

opts

380feda

tests

6dbb284

fix

843421e

test

65ec910

Merge branch 'master' into serdes-refactor

cb5866b

fixes for merge from master

bdf986c

npmignore

1ffef51

malkiz-rapid reviewed Feb 18, 2021

View reviewed changes

package.json Show resolved Hide resolved

malkiz-rapid reviewed Feb 18, 2021

View reviewed changes

package.json Show resolved Hide resolved

dskatz22 mentioned this pull request Feb 21, 2021

upgrading dockest to 2.1.0 #94

Merged

Omri Eyal added 14 commits February 22, 2021 10:00

revert the tests to ensure backwards compatibility

0d2848c

refactor

fe31dcd

backwards compatibility

e2be668

tests

ff622fc

refactor

6550862

rename serdes -> helper

64dbc1f

options types

15a47a6

errors

8986c3d

fix type

86abe1a

revert test

6215464

migrate logic from master src/encoder.ts

0405d6f

lint

e0ad862

patch by @dskatz22 to support protobuf generated by kafka-connect

6895de4

fix options and tests

73e628a

Nevon requested changes Feb 24, 2021

View reviewed changes

Omri Eyal added 3 commits February 24, 2021 15:45

types fix

fa589e8

ProtoOptions

4487fee

errors

a06e3b5

Nevon approved these changes Feb 24, 2021

View reviewed changes

Nevon merged commit 40874b2 into kafkajs:master Feb 24, 2021

This was referenced Feb 24, 2021

Update documentation for major version bump #95

Open

v2.0.0 #96

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

add support for JSON-schema and Protobuf (solves #79) #93

add support for JSON-schema and Protobuf (solves #79) #93

Malkiz commented Feb 16, 2021 •

edited

Loading

malkiz-rapid Feb 18, 2021

malkiz-rapid Feb 18, 2021

Nevon Feb 18, 2021 •

edited

Loading

malkiz-rapid Feb 18, 2021

malkiz-rapid Feb 18, 2021

malkiz-rapid Feb 18, 2021

Nevon Feb 18, 2021

malkiz-rapid Feb 18, 2021

malkiz-rapid Feb 18, 2021

malkiz-rapid Feb 18, 2021

malkiz-rapid Feb 18, 2021

malkiz-rapid Feb 18, 2021

malkiz-rapid Feb 18, 2021

Nevon Feb 24, 2021

Nevon Feb 24, 2021

Nevon Feb 24, 2021

Nevon Feb 24, 2021

malkiz-rapid Feb 24, 2021

Nevon commented Feb 24, 2021

add support for JSON-schema and Protobuf (solves #79) #93

add support for JSON-schema and Protobuf (solves #79) #93

Conversation

Malkiz commented Feb 16, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Nevon Feb 18, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Nevon commented Feb 24, 2021

Malkiz commented Feb 16, 2021 •

edited

Loading

Nevon Feb 18, 2021 •

edited

Loading