feat: add option to specify a version number if schema ID is not avai…

go.mod

@@ -13,6 +13,7 @@ require (
 	github.com/aws/aws-sdk-go-v2/config v1.27.10
 	github.com/aws/aws-sdk-go-v2/credentials v1.17.10
 	github.com/aws/aws-sdk-go-v2/service/kms v1.30.1
+	github.com/aws/aws-sdk-go-v2/service/sts v1.28.6
 	github.com/golang/protobuf v1.5.4
 	github.com/google/cel-go v0.20.1
 	github.com/google/uuid v1.6.0
@@ -57,7 +58,6 @@ require (
 	github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.11.7 // indirect
 	github.com/aws/aws-sdk-go-v2/service/sso v1.20.4 // indirect
 	github.com/aws/aws-sdk-go-v2/service/ssooidc v1.23.4 // indirect
-	github.com/aws/aws-sdk-go-v2/service/sts v1.28.6 // indirect
 	github.com/aws/smithy-go v1.20.2 // indirect
 	github.com/bahlo/generic-list-go v0.2.0 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect

schemaregistry/serde/avrov2/avro.go

@@ -93,6 +93,7 @@ func (s *Serializer) Serialize(topic string, msg interface{}) ([]byte, error) {
 	var err error
 	// Don't derive the schema if it is being looked up in the following ways
 	if s.Conf.UseSchemaID == -1 &&
+		s.Conf.UseSpecificVersion == -1 &&
 		!s.Conf.UseLatestVersion &&
 		len(s.Conf.UseLatestWithMetadata) == 0 {
 		msgType := reflect.TypeOf(msg)

schemaregistry/serde/config.go

@@ -22,6 +22,8 @@ type SerializerConfig struct {
 	AutoRegisterSchemas bool
 	// UseSchemaID specifies a schema ID to use during serialization
 	UseSchemaID int
+	// UseVersion specifies a specific schema version to use during serialization
+	UseSpecificVersion int
 	// UseLatestVersion specifies whether to use the latest schema version during serialization
 	UseLatestVersion bool
 	// UseLatestWithMetadata specifies whether to use the latest schema with metadata during serialization
@@ -38,6 +40,7 @@ func NewSerializerConfig() *SerializerConfig {
 
 	c.AutoRegisterSchemas = true
 	c.UseSchemaID = -1
+	c.UseSpecificVersion = -1
 	c.UseLatestVersion = false
 	c.NormalizeSchemas = false
 

schemaregistry/serde/jsonschema/json_schema.go

@@ -95,6 +95,7 @@ func (s *Serializer) Serialize(topic string, msg interface{}) ([]byte, error) {
 	var err error
 	// Don't derive the schema if it is being looked up in the following ways
 	if s.Conf.UseSchemaID == -1 &&
+		s.Conf.UseSpecificVersion == -1 &&
 		!s.Conf.UseLatestVersion &&
 		len(s.Conf.UseLatestWithMetadata) == 0 {
 		jschema := jsonschema.Reflect(msg)

schemaregistry/serde/protobuf/protobuf.go

@@ -208,6 +208,7 @@ func (s *Serializer) Serialize(topic string, msg interface{}) ([]byte, error) {
 	var err error
 	// Don't derive the schema if it is being looked up in the following ways
 	if s.Conf.UseSchemaID == -1 &&
+		s.Conf.UseSpecificVersion == -1 &&
 		!s.Conf.UseLatestVersion &&
 		len(s.Conf.UseLatestWithMetadata) == 0 {
 		schemaInfo, err := s.getSchemaInfo(protoMsg)

schemaregistry/serde/serde.go

@@ -408,6 +408,7 @@ func TopicNameStrategy(topic string, serdeType Type, schema schemaregistry.Schem
 func (s *BaseSerializer) GetID(topic string, msg interface{}, info *schemaregistry.SchemaInfo) (int, error) {
 	autoRegister := s.Conf.AutoRegisterSchemas
 	useSchemaID := s.Conf.UseSchemaID
+	useSpecificVersion := s.Conf.UseSpecificVersion
 	useLatestWithMetadata := s.Conf.UseLatestWithMetadata
 	useLatest := s.Conf.UseLatestVersion
 	normalizeSchema := s.Conf.NormalizeSchemas
@@ -428,6 +429,13 @@ func (s *BaseSerializer) GetID(topic string, msg interface{}, info *schemaregist
 			return -1, err
 		}
 		id = useSchemaID
+	} else if useSpecificVersion >= 0 {
+		metadata, err := s.Client.GetSchemaMetadata(subject, useSpecificVersion)
+		if err != nil {
+			return -1, err
+		}
+		*info = metadata.SchemaInfo
+		id = metadata.ID
 	} else if len(useLatestWithMetadata) != 0 {
 		metadata, err := s.Client.GetLatestWithMetadata(subject, useLatestWithMetadata, true)
 		if err != nil {