Merge pull request #49 from dfds/feature/2471-make-version-part-of-co…

…nfluent-cloud-schema-registry-registering-request added schema version part of request to confluent cloud
dfds · Jan 11, 2024 · 1d7fc60 · 1d7fc60
2 parents 1764a4d + c0b6517
commit 1d7fc60
Show file tree

Hide file tree

Showing 10 changed files with 20 additions and 6 deletions.
diff --git a/db/migrations/20240110130102_add_schema_version_column.sql b/db/migrations/20240110130102_add_schema_version_column.sql
@@ -0,0 +1,4 @@
+-- 2024-01-10 13:01:02 : add schema version
+
+ALTER TABLE schema_process
+    ADD COLUMN schema_version int NOT NULL DEFAULT 1;
diff --git a/src/functional_tests/create_schema_test.go b/src/functional_tests/create_schema_test.go
@@ -20,10 +20,12 @@ func setupCreateSchemaHttpMock(processInput schema.ProcessInput, topicName strin
 	type schemaPayload struct {
 		SchemaType string `json:"schemaType"`
 		Schema     string `json:"schema"`
+		Version    int32  `json:"version"`
 	}
 	payload, err := json.Marshal(schemaPayload{
 		SchemaType: "JSON",
 		Schema:     processInput.Schema,
+		Version:    processInput.SchemaVersion,
 	})
 	if err != nil {
 		panic(err)
@@ -100,6 +102,7 @@ func TestCreateSchemaProcess(t *testing.T) {
 		MessageType:       "message-type-for-schema-registry",
 		Description:       "schema-description",
 		Schema:            "test-schema",
+		SchemaVersion:     1,
 	}
 
 	setupCreateSchemaHttpMock(input, createSchemaVariables.TopicName, testerApp.dbSeedVariables)

diff --git a/src/internal/confluent/client.go b/src/internal/confluent/client.go
@@ -461,11 +461,12 @@ func (c *Client) delete(ctx context.Context, url string, apiKey models.ApiKey) (
 }
 
 type schemaPayload struct {
+	Version    int32  `json:"version"`
 	SchemaType string `json:"schemaType"`
 	Schema     string `json:"schema"`
 }
 
-func (c *Client) RegisterSchema(ctx context.Context, clusterId models.ClusterId, subject string, schema string) error {
+func (c *Client) RegisterSchema(ctx context.Context, clusterId models.ClusterId, subject string, schema string, version int32) error {
 	cluster, err := c.clusters.Get(clusterId)
 
 	if err != nil {
@@ -479,6 +480,7 @@ func (c *Client) RegisterSchema(ctx context.Context, clusterId models.ClusterId,
 	url := fmt.Sprintf("%s/subjects/%s/versions", cluster.SchemaRegistryApiEndpoint, subject)
 
 	payload, err := json.Marshal(schemaPayload{
+		Version:    version,
 		SchemaType: "JSON",
 		Schema:     schema,
 	})

diff --git a/src/internal/models/schema.go b/src/internal/models/schema.go
@@ -16,9 +16,10 @@ type SchemaProcess struct {
 	Schema            string
 	CreatedAt         time.Time
 	CompletedAt       *time.Time
+	SchemaVersion     int32
 }
 
-func NewSchemaProcess(clusterId ClusterId, messageContractId string, topicId string, messageType string, description string, subject string, schema string) *SchemaProcess {
+func NewSchemaProcess(clusterId ClusterId, messageContractId string, topicId string, messageType string, description string, subject string, schema string, schemaVersion int32) *SchemaProcess {
 	return &SchemaProcess{
 		Id:                uuid.NewV4(),
 		ClusterId:         clusterId,
@@ -30,6 +31,7 @@ func NewSchemaProcess(clusterId ClusterId, messageContractId string, topicId str
 		Schema:            schema,
 		CreatedAt:         time.Now(),
 		CompletedAt:       nil,
+		SchemaVersion:     schemaVersion,
 	}
 }
 

diff --git a/src/internal/schema/confluent.go b/src/internal/schema/confluent.go
@@ -12,5 +12,5 @@ type Confluent interface {
 	CreateServiceAccountRoleBinding(ctx context.Context, serviceAccount models.ServiceAccountId, clusterId models.ClusterId) error
 	CountSchemaRegistryApiKeys(ctx context.Context, clusterAccess models.ServiceAccountId, clusterId models.ClusterId) (int, error)
 	DeleteSchemaRegistryApiKey(ctx context.Context, clusterId models.ClusterId, serviceAccountId models.ServiceAccountId) error
-	RegisterSchema(ctx context.Context, clusterId models.ClusterId, subject string, schema string) error
+	RegisterSchema(ctx context.Context, clusterId models.ClusterId, subject string, schema string, version int32) error
 }
diff --git a/src/internal/schema/context.go b/src/internal/schema/context.go
@@ -51,7 +51,7 @@ func (c *StepContext) IsCompleted() bool {
 }
 
 func (c *StepContext) RegisterSchema() error {
-	return c.registry.RegisterSchema(c.ctx, c.state.ClusterId, c.state.Subject, c.state.Schema)
+	return c.registry.RegisterSchema(c.ctx, c.state.ClusterId, c.state.Subject, c.state.Schema, c.state.SchemaVersion)
 }
 
 func (c *StepContext) MarkAsCompleted() {

diff --git a/src/internal/schema/handler.go b/src/internal/schema/handler.go
@@ -28,6 +28,7 @@ func (h *handler) Handle(ctx context.Context, msgContext messaging.MessageContex
 			MessageType:       message.MessageType,
 			Description:       message.Description,
 			Schema:            message.Schema,
+			SchemaVersion:     message.SchemaVersion,
 		}
 		return h.process.Process(ctx, input)
 

diff --git a/src/internal/schema/message.go b/src/internal/schema/message.go
@@ -6,6 +6,7 @@ type MessageContractRequested struct {
 	MessageType       string `json:"messageType"`
 	Description       string `json:"description"`
 	Schema            string `json:"schema"`
+	SchemaVersion     int32  `json:"schemaVersion"`
 }
 
 type SchemaRegistered struct {

diff --git a/src/internal/schema/process.go b/src/internal/schema/process.go
@@ -42,6 +42,7 @@ type ProcessInput struct {
 	MessageType       string
 	Description       string
 	Schema            string
+	SchemaVersion     int32
 }
 
 func (p *process) Process(ctx context.Context, input ProcessInput) error {
@@ -123,7 +124,7 @@ func getOrCreateProcessState(repo schemaRepository, input ProcessInput, topic *m
 	}
 
 	subject := fmt.Sprintf("%s-%s", topic.Name, input.MessageType)
-	schema = models.NewSchemaProcess(topic.ClusterId, input.MessageContractId, input.TopicId, input.MessageType, input.Description, subject, input.Schema)
+	schema = models.NewSchemaProcess(topic.ClusterId, input.MessageContractId, input.TopicId, input.MessageType, input.Description, subject, input.Schema, input.SchemaVersion)
 
 	if err := repo.SaveSchemaProcessState(schema); err != nil {
 		return nil, err

diff --git a/src/internal/schema/registry.go b/src/internal/schema/registry.go
@@ -6,5 +6,5 @@ import (
 )
 
 type SchemaRegistry interface {
-	RegisterSchema(ctx context.Context, clusterId models.ClusterId, subject string, schema string) error
+	RegisterSchema(ctx context.Context, clusterId models.ClusterId, subject string, schema string, version int32) error
 }