protobuf validation failure fix (#1292)

Author: ankitk-me

runtime/binding-kafka/src/main/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/KafkaCacheClientProduceFactory.java

@@ -760,7 +760,7 @@ private void onClientInitialData(
                 flushClientFanInitialIfNecessary(traceId);
             }
 
-            if ((flags & FLAGS_INCOMPLETE) != 0x00)
+            if ((flags & FLAGS_INCOMPLETE) != 0x00 || error == ERROR_INVALID_RECORD)
             {
                 markEntryDirty(traceId, stream.partitionOffset);
             }

runtime/model-protobuf/src/main/java/io/aklivity/zilla/runtime/model/protobuf/internal/ProtobufReadConverterHandler.java

@@ -188,7 +188,6 @@ private int validate(
                 try
                 {
                     DynamicMessage message = builder.mergeFrom(in).build();
-                    builder.clear();
                     if (!message.getUnknownFields().asMap().isEmpty())
                     {
                         break validate;
@@ -215,6 +214,10 @@ private int validate(
                 {
                     event.validationFailure(traceId, bindingId, ex.getMessage());
                 }
+                finally
+                {
+                    builder.clear();
+                }
             }
         }
         return valLength;

runtime/model-protobuf/src/main/java/io/aklivity/zilla/runtime/model/protobuf/internal/ProtobufWriteConverterHandler.java

@@ -111,13 +111,16 @@ private boolean validate(
                 try
                 {
                     DynamicMessage message = builder.mergeFrom(in).build();
-                    builder.clear();
                     status = message.getUnknownFields().asMap().isEmpty();
                 }
                 catch (IOException ex)
                 {
                     event.validationFailure(traceId, bindingId, ex.getMessage());
                 }
+                finally
+                {
+                    builder.clear();
+                }
             }
         }
         return status;
@@ -175,7 +178,6 @@ private int serializeJsonRecord(
                 {
                     parser.merge(input, builder);
                     DynamicMessage message = builder.build();
-                    builder.clear();
                     if (message.isInitialized() && message.getUnknownFields().asMap().isEmpty())
                     {
                         out.wrap(out.buffer());
@@ -187,6 +189,10 @@ private int serializeJsonRecord(
                 {
                     event.validationFailure(traceId, bindingId, ex.getMessage());
                 }
+                finally
+                {
+                    builder.clear();
+                }
             }
         }
         return valLength;