Fix incorrect calculation of canonical schema

canonical.go

@@ -120,6 +120,22 @@ func pcfObject(jsonMap map[string]interface{}, parentNamespace string, typeLooku
 			}
 		}
 
+		// This fixes a fairly subtle bug that can occur during schema canonicalization,
+		// which can randomly namespace a name that should not be, depending on the
+		// order of map traversal.
+		if theType, ok := jsonMap["type"]; ok {
+			// Check it's not a type that should be namespaced in canonical form, i.e.
+			// this should be map[string]interface{} and not "record".
+			if _, ook := theType.(string); !ook {
+				if valStr, oook := v.(string); oook {
+					// if we're an interface{} type which has an entry in typeLookup,
+					// remove it before we recurse into parsingCanonicalForm, as
+					// the wrong namespace will be applied.
+					delete(typeLookup, valStr)
+				}
+			}
+		}
+
 		pk, err := parsingCanonicalForm(k, parentNamespace, typeLookup)
 		if err != nil {
 			return "", err

canonical_test.go

@@ -285,3 +285,44 @@ func TestCanonicalSchema(t *testing.T) {
 		}
 	}
 }
+
+func TestCanonicalSchemaFingerprintFlake(t *testing.T) {
+	const schema = `{
+  "name" : "thename",
+  "type" : "record",
+  "namespace" : "com.fooname",
+  "fields" : [
+    {
+      "name" : "bar" ,
+      "type" : {
+        "name" : "bar",
+        "type" : "record",
+        "fields" : [
+          {
+            "name" : "car",
+            "type" : "int"
+          }
+        ]
+      }
+    }
+  ]
+}
+`
+	codec, err := NewCodec(schema)
+	if err != nil {
+		t.Fatalf("unexpected schema parse failure, err=%v", err)
+	}
+	prevRun := codec.Rabin
+
+	// This test is flaky. It exposes a bug that manifests due to
+	// randomized map iteration. However, 32 iterations should give
+	// high probability to see the failure.
+	for i := 0; i < 32; i++ {
+		codec, err = NewCodec(schema)
+		currentRun := codec.Rabin
+		if prevRun != currentRun {
+			t.Fatalf("same schema should always have same fingerprint, rabin1: %d, rabin2: %d", prevRun, currentRun)
+		}
+		prevRun = currentRun
+	}
+}