Avro serialization junk value in confluent cloud

i am using this client to create the schema and send avro messages to topic. This works fine in my local machine ( macOS ) and in my GCP compute unit where downloaded version of confluent is running.

After the trails i got the confluent cloud access and tried with basic avro serialization doesnt seems to work as expected. The value section is always prefixed with junk characters and doesn't have the same behaviour which worked out in my local machine and my GCP single machine.

Here is the details and code snippet and the process which is followed.

Created a topic in confluent cloud by name sample_topic_003

Schema File for Avro

{
"doc": "Sample schema to help you get started...",
"fields": [
  {
    "doc": "The int type is a 32-bit signed integer.",
    "name": "id",
    "type": "int"
  },
  {
    "doc": "The string is a unicode character sequence.",
    "name": "name",
    "type": "string"
  }
],
"name": "my_sample_schema_three",
"namespace": "com.mycorp.mynamespace.local.three",
"type": "record"
}

Golang Code snippet for sending the data to confluent cloud

func MethodPOCAvro() {

    type ComplexType struct {
        ID   int    `json:"id"`
        Name string `json:"name"`
    }

    sampleSchema := `{
        "doc": "Sample schema to help you get started...",
        "fields": [
          {
            "doc": "The int type is a 32-bit signed integer.",
            "name": "id",
            "type": "int"
          },
          {
            "doc": "The string is a unicode character sequence.",
            "name": "name",
            "type": "string"
          }
        ],
        "name": "my_sample_schema_three",
        "namespace": "com.mycorp.mynamespace.local.three",
        "type": "record"
      }`
    topic := "sample_topic_003"
    kafkaSchemaRegistryURL = "https://xxx.gcp.confluent.cloud"

    /*** Start registering schemas in the confluent schema registry ***/
    schemaRegistryClient := srclient.CreateSchemaRegistryClient(kafkaSchemaRegistryURL)
    schemaRegistryClient.SetCredentials("xxx", "xxx")

    schema, err := schemaRegistryClient.GetLatestSchema(topic, false)
    if schema == nil {
        log.Print("Schema not found registering now ")
        schema, err = schemaRegistryClient.CreateSchema(topic, sampleSchema, srclient.Avro, false)
        if schema != nil {
            log.Print("Register successful ")
        }
        if err != nil {
            panic(fmt.Sprintf("Error creating the schema  %s", err))
        }
    }

    // Init producer
    p, err := kafka.NewProducer(&kafka.ConfigMap{
        "bootstrap.servers": "xxx.confluent.cloud:9092",
        "sasl.mechanisms":   "PLAIN",
        "security.protocol": "SASL_SSL",
        // "debug":                               "all",
        "sasl.username":                       "xxx",
        "enable.ssl.certificate.verification": "false",
        "sasl.password":                       "xxx"})

        time.Sleep(1 * time.Second)

        schemaIDBytes := make([]byte, 4)
        binary.BigEndian.PutUint32(schemaIDBytes, uint32(schema.ID()))
        log.Println("Schema ID ", schema.ID())

        // 3) Serialize the record using the schema provided by the client,
        // making sure to include the schema id as part of the record.
        newComplexType := ComplexType{ID: 1, Name: "Gopher"}
        value, _ := json.Marshal(newComplexType)
        log.Println("value ", value)

        native, _, _ := schema.Codec().NativeFromTextual(value)
        log.Println("native ", native)

        valueBytes, _ := schema.Codec().BinaryFromNative(nil, native)
        log.Println("valueBytes ", valueBytes)
        var recordValue []byte
        recordValue = append(recordValue, byte(0))
        recordValue = append(recordValue, schemaIDBytes...)
        recordValue = append(recordValue, valueBytes...)
        key, _ := uuid.NewUUID()
        log.Println("recordValue ", recordValue)

        err = p.Produce(&kafka.Message{
            TopicPartition: kafka.TopicPartition{
                Topic: &topic, Partition: kafka.PartitionAny},
            Key: []byte(key.String()), Value: recordValue}, nil)

        if err != nil {
            fmt.Fprintf(os.Stderr, "Exception while producing the message .\n")
        }

        k := p.Flush(200)
        log.Print("Pushed ", k)

}

After the program execution I am expecting value.id and value.name but i get only value having prefixed with junk character.

I have even verified the bytes being passed on to confluent cloud. for the above program following the console output

Schema not found registering now 
Register successful 

Schema ID  100022
value  [123 34 105 100 34 58 49 44 34 110 97 109 101 34 58 34 71 111 112 104 101 114 34 125]
native  map[id:1 name:Gopher]
valueBytes  [2 12 71 111 112 104 101 114]
recordValue  [0 0 1 134 182 2 12 71 111 112 104 101 114]
Pushed 1

I believe it is as per confluent wire order doc where first byte is 0 and then 1-4 bytes are of schema ID and then the actual message. ( https://docs.confluent.io/current/schema-registry/serdes-develop/index.html#wire-format )
The same program works for in my local setup ( macOX ) and in my GCP where confluent is installed. following is the screenshot of the same.
The output of my client program when ran in my local setup .
Console output of my local setup

Schema ID  1
value  [123 34 105 100 34 58 49 44 34 110 97 109 101 34 58 34 71 111 112 104 101 114 34 125]
native  map[id:1 name:Gopher]
valueBytes  [2 12 71 111 112 104 101 114]
recordValue  [0 0 0 0 1 2 12 71 111 112 104 101 114]
Pushed 1

Notes :

If you notice from 1st till the 4th byte there is variation due to schema ID.
Schema ID returned from my local setup is single digit but in confluent cloud i got schema ID of 6 digits.
Do I need to consider any headers while generating wire order here. ?

riferrei / srclient

Avro serialization junk value in confluent cloud #15