From 91cf77cbed25bd3a92f153503950d7047f9700b3 Mon Sep 17 00:00:00 2001
From: Mostafa Moradian <mostafamoradian0@gmail.com>
Date: Thu, 4 Aug 2022 19:41:49 +0200
Subject: [PATCH] Fix regression in JSONSchema (de)serialization

Fix checking headers in the script
Remove unused error code and recorder error codes
---
 error_codes.go                                |  5 ++--
 jsonschema.go                                 | 27 ++++++++++++-------
 .../test_jsonschema_with_schema_registry.js   |  2 +-
 3 files changed, 20 insertions(+), 14 deletions(-)

diff --git a/error_codes.go b/error_codes.go
index c9121f6..db19adc 100644
--- a/error_codes.go
+++ b/error_codes.go
@@ -29,9 +29,8 @@ const (
 	failedUnmarshalJSON         errCode = 2007
 	failedValidateJSON          errCode = 2008
 	failedEncodeToJSON          errCode = 2009
-	failedEncodeJSONToBinary    errCode = 2010
-	failedDecodeJSONFromBinary  errCode = 2011
-	failedToUnmarshalSchema     errCode = 2012
+	failedDecodeJSONFromBinary  errCode = 2010
+	failedToUnmarshalSchema     errCode = 2011
 
 	// producer.
 	failedWriteMessage errCode = 3000
diff --git a/jsonschema.go b/jsonschema.go
index 5b6d789..7a06b06 100644
--- a/jsonschema.go
+++ b/jsonschema.go
@@ -71,18 +71,18 @@ func SerializeJSON(
 	if schemaInfo != nil {
 		schemaID = schemaInfo.ID()
 
-		// Encode the data into JSON and then the wire format
-		jsonEncodedData, _, err := schemaInfo.Codec().NativeFromTextual(bytesData)
-		if err != nil {
-			return nil, NewXk6KafkaError(failedEncodeToJSON,
-				"Failed to encode data into JSON",
+		var jsonBytes interface{}
+		if err := json.Unmarshal(bytesData, &jsonBytes); err != nil {
+			return nil, NewXk6KafkaError(failedUnmarshalJSON,
+				"Failed to unmarshal JSON data",
 				err)
 		}
 
-		bytesData, err = schemaInfo.Codec().BinaryFromNative(nil, jsonEncodedData)
+		// Encode the data into JSON and then the wire format
+		err := schemaInfo.JsonSchema().Validate(jsonBytes)
 		if err != nil {
-			return nil, NewXk6KafkaError(failedEncodeJSONToBinary,
-				"Failed to encode JSON data into binary",
+			return nil, NewXk6KafkaError(failedEncodeToJSON,
+				"Failed to encode data into JSON",
 				err)
 		}
 	}
@@ -148,14 +148,21 @@ func DeserializeJSON(
 	}
 
 	if schemaInfo != nil {
+		var jsonBytes interface{}
+		if err := json.Unmarshal(bytesDecodedData, &jsonBytes); err != nil {
+			return nil, NewXk6KafkaError(failedUnmarshalJSON,
+				"Failed to unmarshal JSON data",
+				err)
+		}
+
 		// Decode the data from Json
-		jsonDecodedData, _, err := schemaInfo.Codec().NativeFromBinary(bytesDecodedData)
+		err := schemaInfo.JsonSchema().Validate(jsonBytes)
 		if err != nil {
 			return nil, NewXk6KafkaError(failedDecodeJSONFromBinary,
 				"Failed to decode data from JSON",
 				err)
 		}
-		return jsonDecodedData, nil
+		return jsonBytes, nil
 	}
 
 	return bytesDecodedData, nil
diff --git a/scripts/test_jsonschema_with_schema_registry.js b/scripts/test_jsonschema_with_schema_registry.js
index 69e626e..b578c80 100644
--- a/scripts/test_jsonschema_with_schema_registry.js
+++ b/scripts/test_jsonschema_with_schema_registry.js
@@ -108,7 +108,7 @@ export default function () {
         "Key is correct": (msg) => msg.key.key == "key0",
         "Value is correct": (msg) =>
             msg.value.firstName == "firstName-0" && msg.value.lastName == "lastName-0",
-        "Headers are correct": (msg) => msg.headers.length == 0,
+        "Headers are correct": (msg) => msg.headers.hasOwnProperty("mykey") == false,
         "Time is past": (msg) => new Date(msg["time"]) < new Date(),
         "Offset is correct": (msg) => msg.offset == 0,
         "Partition is correct": (msg) => msg.partition == 0,