Set skipKnownTypes default to be consistent with other clients (#2376)

* Change skipKnownTypes to be consistent with other clients

* Enhance test

Author: rayokota

src/Confluent.SchemaRegistry.Serdes.Protobuf/ProtobufSerializer.cs

@@ -54,7 +54,7 @@ namespace Confluent.SchemaRegistry.Serdes
     /// </remarks>
     public class ProtobufSerializer<T> : AsyncSerializer<T, FileDescriptorSet>  where T : IMessage<T>, new()
     {
-        private bool skipKnownTypes;
+        private bool skipKnownTypes = true;
         private bool useDeprecatedFormat;
         private ReferenceSubjectNameStrategyDelegate referenceSubjectNameStrategy;
 
@@ -190,7 +190,7 @@ private async Task<List<SchemaReference>> RegisterOrGetReferences(FileDescriptor
             for (int i=0; i<fd.Dependencies.Count; ++i)
             {
                 FileDescriptor fileDescriptor = fd.Dependencies[i];
-                if (skipKnownTypes && fileDescriptor.Name.StartsWith("google/protobuf/"))
+                if (skipKnownTypes && IgnoreReference(fileDescriptor.Name))
                 {
                     continue;
                 }
@@ -328,5 +328,12 @@ protected override async Task<FileDescriptorSet> ParseSchema(Schema schema)
                 .ConfigureAwait(continueOnCapturedContext: false);
             return ProtobufUtils.Parse(schema.SchemaString, references);
         }
+
+        protected override bool IgnoreReference(string name)
+        {
+            return name.StartsWith("confluent/") ||
+                   name.StartsWith("google/protobuf/") ||
+                   name.StartsWith("google/type/");
+        }
     }
 }

src/Confluent.SchemaRegistry.Serdes.Protobuf/ProtobufSerializerConfig.cs

@@ -195,7 +195,7 @@ public IDictionary<string, string> UseLatestWithMetadata
         ///     Specifies whether or not the Protobuf serializer should skip known types
         ///     when resolving dependencies.
         ///
-        ///     default: false
+        ///     default: true
         /// </summary>
         public bool? SkipKnownTypes
         {

src/Confluent.SchemaRegistry/AsyncSerde.cs

@@ -142,14 +142,19 @@ protected async Task<IDictionary<string, string>> ResolveReferences(Schema schem
                 .ConfigureAwait(continueOnCapturedContext: false);
             return result;
         }
+
+        protected virtual bool IgnoreReference(string name)
+        {
+            return false;
+        }
 
         private async Task<IDictionary<string, string>> ResolveReferences(
             Schema schema, IDictionary<string, string> schemas, ISet<string> visited)
         {
             IList<SchemaReference> references = schema.References;
             foreach (SchemaReference reference in references)
             {
-                if (visited.Contains(reference.Name))
+                if (IgnoreReference(reference.Name) || visited.Contains(reference.Name))
                 {
                     continue;
                 }

test/Confluent.SchemaRegistry.Serdes.IntegrationTests/Tests_Protobuf/ProduceConsumeGoogleRef.cs

@@ -33,12 +33,13 @@ public static void ProduceConsumeGoogleRefProtobuf(string bootstrapServers, stri
         {
             var producerConfig = new ProducerConfig { BootstrapServers = bootstrapServers };
             var schemaRegistryConfig = new SchemaRegistryConfig { Url = schemaRegistryServers };
+            var serializerConfig = new ProtobufSerializerConfig() { SkipKnownTypes = false };
 
             using (var topic = new TemporaryTopic(bootstrapServers, 1))
             using (var schemaRegistry = new CachedSchemaRegistryClient(schemaRegistryConfig))
             using (var producer =
                 new ProducerBuilder<string, WithGoogleRefs.TheRecord>(producerConfig)
-                    .SetValueSerializer(new ProtobufSerializer<WithGoogleRefs.TheRecord>(schemaRegistry))
+                    .SetValueSerializer(new ProtobufSerializer<WithGoogleRefs.TheRecord>(schemaRegistry, serializerConfig))
                     .Build())
             {
                 var u = new WithGoogleRefs.TheRecord();