schemaregistry

README

Schema Registry CLI and client

This repository contains a Command Line Interface (CLI) and a Go client for the REST API of Confluent's Kafka Schema Registry.

CLI

To install the CLI, assuming a properly setup Go installation, do:

go get -u github.com/landoop/schema-registry/schema-registry-cli

After that, the CLI is found in $GOPATH/bin/schema-registry-cli. Running schema-registry-cli without arguments gives:

A command line interface for the Confluent schema registry

Usage:
  schema-registry-cli [command]

Available Commands:
  add         registers the schema provided through stdin
  compatible  tests compatibility between a schema from stdin and a given subject
  exists      checks if the schema provided through stdin exists for the subject
  get         retrieves a schema specified by id or subject
  get-config  retrieves global or suject specific configuration
  subjects    lists all registered subjects
  versions    lists all available versions

Flags:
  -h, --help         help for schema-registry-cli
  -n, --no-color     dont color output
  -e, --url string   schema registry url, overrides SCHEMA_REGISTRY_URL (default "http://localhost:8081")
  -v, --verbose      be verbose

Use "schema-registry-cli [command] --help" for more information about a command.

The schema registry url can be configured through the SCHEMA_REGISTRY_URL environment variable, and overridden through --url. When none is provided, http://localhost:8081 is used as default.

Client

The client package provides a client to deal with the registry from code. It is used by the CLI internally. Usage looks like:

import "github.com/landoop/schema-registry"

client, _ := schemaregistry.NewClient(schemaregistry.DefaultUrl)
client.Subjects()

Or, to use with a Schema Registry endpoint listening on HTTPS:

import (
    "crypto/tls"
    "crypto/x509"
    "io/ioutil"

    "github.com/landoop/schema-registry"
)

// Create a TLS config to use to connect to Schema Registry. This config will permit TLS connections to an endpoint
// whose TLS cert is signed by the given caFile.
caCert, err := ioutil.ReadFile("/path/to/ca/file")
if err != nil {
    panic(err)
}

caCertPool := x509.NewCertPool()
caCertPool.AppendCertsFromPEM(caCert)

tlsConfig :=  &tls.Config{
    RootCAs:            caCertPool,
    InsecureSkipVerify: true,
}

httpsClientTransport := &http.Transport{
  TLSClientConfig: tlsConfig,
}

httpsClient := &http.Client{
  Transport: httpsClientTransport,
}

// Create the Schema Registry client
client, _ := schemaregistry.NewClient(baseurl, UsingClient(httpsClient))
client.Subjects()

The documentation of the package can be found here:

Documentation

Overview

Package schemaregistry provides a client for Confluent's Kafka Schema Registry REST API.

Index

• Constants
• func IsSchemaNotFound(err error) bool
• func IsSubjectNotFound(err error) bool
• func JSONAvroSchema(avroSchema string) (json.RawMessage, error)
• type Client
  • func NewClient(baseURL string, options ...Option) (*Client, error)
  • func NewTlsClient(baseurl string, tlsConfig *tls.Config) (*Client, error)
  • func (c *Client) DeleteSubject(subject string) (versions []int, err error)
  • func (c *Client) GetConfig(subject string) (Config, error)
  • func (c *Client) GetLatestSchema(subject string) (Schema, error)
  • func (c *Client) GetSchemaByID(subjectID int) (string, error)
  • func (c *Client) GetSchemaById(id int) (string, error)
  • func (c *Client) GetSchemaBySubject(subject string, versionID int) (Schema, error)
  • func (c *Client) IsLatestSchemaCompatible(subject string, avroSchema string) (bool, error)
  • func (c *Client) IsRegistered(subject, schema string) (bool, Schema, error)
  • func (c *Client) IsSchemaCompatible(subject string, avroSchema string, versionID int) (bool, error)
  • func (c *Client) RegisterNewSchema(subject string, avroSchema string) (int, error)
  • func (c *Client) Subjects() (subjects []string, err error)
  • func (c *Client) Versions(subject string) (versions []int, err error)
• type Config
• type Option
  • func UsingClient(httpClient *http.Client) Option
• type ResourceError
  • func (err ResourceError) Error() string
• type Schema

Constants

const DefaultURL = "http://localhost:8081"

DefaultURL is the address where a local schema registry listens by default.

const DefaultUrl = DefaultURL

DefaultUrl is the address where a local schema registry listens by default.

DEPRECATED: Use `schemaregistry.DefaultURL` instead.

const SchemaLatestVersion = "latest"

SchemaLatestVersion is the only one valid string for the "versionID", it's the "latest" version string and it's used on `GetLatestSchema`.

Functions

func IsSchemaNotFound

func IsSchemaNotFound(err error) bool

IsSchemaNotFound checks the returned error to see if it is kind of a schema not found error code.

func IsSubjectNotFound

func IsSubjectNotFound(err error) bool

IsSubjectNotFound checks the returned error to see if it is kind of a subject not found error code.

func JSONAvroSchema

func JSONAvroSchema(avroSchema string) (json.RawMessage, error)

JSONAvroSchema converts and returns the json form of the "avroSchema" as []byte.

Types

type Client

type Client struct {
	// contains filtered or unexported fields
}

Client is the registry schema REST API client.

func NewClient

func NewClient(baseURL string, options ...Option) (*Client, error)

NewClient creates & returns a new Registry Schema Client based on the passed url and the options.

func NewTlsClient

func NewTlsClient(baseurl string, tlsConfig *tls.Config) (*Client, error)

NewTlsClient returns a new Client that securely connects to baseurl.

DEPRECATED: Use `schemaregistry.NewClient(baseURL, schemaregistry.UsingClient(customHTTPSClient))` instead.

func (*Client) DeleteSubject

func (c *Client) DeleteSubject(subject string) (versions []int, err error)

DeleteSubject deletes the specified subject and its associated compatibility level if registered. It is recommended to use this API only when a topic needs to be recycled or in development environment. Returns the versions of the schema deleted under this subject.

func (*Client) GetConfig

func (c *Client) GetConfig(subject string) (Config, error)

GetConfig returns the configuration (Config type) for global Schema-Registry or a specific subject. When Config returned has "compatibilityLevel" empty, it's using global settings.

func (*Client) GetLatestSchema

func (c *Client) GetLatestSchema(subject string) (Schema, error)

GetLatestSchema returns the latest version of a schema. See `GetSchemaAtVersion` to retrieve a subject schema by a specific version.

func (*Client) GetSchemaByID

func (c *Client) GetSchemaByID(subjectID int) (string, error)

GetSchemaByID returns the Auro schema string identified by the id. id (int) – the globally unique identifier of the schema.

func (*Client) GetSchemaById

func (c *Client) GetSchemaById(id int) (string, error)

GetSchemaById returns the schema for some id. The schema registry only provides the schema itself, not the id, subject or version.

DEPRECATED: Use `Client#GetSchemaByID` instead.

func (*Client) GetSchemaBySubject

func (c *Client) GetSchemaBySubject(subject string, versionID int) (Schema, error)

GetSchemaBySubject returns the schema for a particular subject and version.

func (*Client) IsLatestSchemaCompatible

func (c *Client) IsLatestSchemaCompatible(subject string, avroSchema string) (bool, error)

IsLatestSchemaCompatible tests compatibility with the latest version of a subject's schema.

func (*Client) IsRegistered

func (c *Client) IsRegistered(subject, schema string) (bool, Schema, error)

IsRegistered tells if the given "schema" is registered for this "subject".

func (*Client) IsSchemaCompatible

func (c *Client) IsSchemaCompatible(subject string, avroSchema string, versionID int) (bool, error)

IsSchemaCompatible tests compatibility with a specific version of a subject's schema.

func (*Client) RegisterNewSchema

func (c *Client) RegisterNewSchema(subject string, avroSchema string) (int, error)

RegisterNewSchema registers a schema. The returned identifier should be used to retrieve this schema from the schemas resource and is different from the schema’s version which is associated with that name.

func (*Client) Subjects

func (c *Client) Subjects() (subjects []string, err error)

Subjects returns a list of the available subjects(schemas). https://docs.confluent.io/current/schema-registry/docs/api.html#subjects

func (*Client) Versions

func (c *Client) Versions(subject string) (versions []int, err error)

Versions returns all schema version numbers registered for this subject.

type Config

type Config struct {
	// CompatibilityLevel mode of subject or global
	CompatibilityLevel string `json:"compatibilityLevel"`
}

Config describes a subject or globa schema-registry configuration

type Option

type Option func(*Client)

Option describes an optional runtime configurator that can be passed on `NewClient`. Custom `Option` can be used as well, it's just a type of `func(*schemaregistry.Client)`.

Look `UsingClient`.

func UsingClient

func UsingClient(httpClient *http.Client) Option

UsingClient modifies the underline HTTP Client that schema registry is using for contact with the backend server.

type ResourceError

type ResourceError struct {
	ErrorCode int    `json:"error_code"`
	Method    string `json:"method,omitempty"`
	URI       string `json:"uri,omitempty"`
	Message   string `json:"message,omitempty"`
}

ResourceError is being fired from all API calls when an error code is received.

func (ResourceError) Error

func (err ResourceError) Error() string

type Schema

type Schema struct {
	// Schema is the Avro schema string.
	Schema string `json:"schema"`
	// Subject where the schema is registered for.
	Subject string `json:"subject"`
	// Version of the returned schema.
	Version int `json:"version"`
	ID      int `json:"id,omitempty"`
}

Schema describes a schema, look `GetSchema` for more.