Adjust padding to accommodate good enough headers and don't include p…

…artial data frame while computing crc32c value

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

@@ -86,7 +86,7 @@ public final class KafkaClientProduceFactory extends KafkaClientSaslHandshaker i
 {
     private static final int PRODUCE_REQUEST_RECORDS_OFFSET_MAX = 512;
 
-    private static final int KAFKA_RECORD_FRAMING = 100; // TODO
+    private static final int KAFKA_RECORD_FRAMING = 512; // TODO
 
     private static final int FLAGS_CON = 0x00;
     private static final int FLAGS_FIN = 0x01;
@@ -539,6 +539,7 @@ private int flushRecordInit(
         final int valueSize = payload != null ? payload.sizeof() : 0;
         client.valueCompleteSize = valueSize + client.encodeableRecordBytesDeferred;
 
+
         final int maxEncodeableBytes = client.encodeSlotLimit + client.valueCompleteSize + KAFKA_RECORD_FRAMING;
         if (client.encodeSlot != NO_SLOT &&
             maxEncodeableBytes > encodePool.slotCapacity())
@@ -1191,6 +1192,7 @@ private final class KafkaProduceClient extends KafkaSaslClient
             private int encodeableRecordCount;
             private int encodeableRecordBytes;
             private int encodeableRecordBytesDeferred;
+            private int encodeableRecordValueBytes;
             private int flushableRequestBytes;
 
             private int decodeSlot = NO_SLOT;
@@ -1652,6 +1654,7 @@ private void doEncodeRecordInit(
 
                 encodeSlotBuffer.putBytes(encodeSlotLimit, encodeBuffer, 0, encodeProgress);
                 encodeSlotLimit += encodeProgress;
+                encodeableRecordValueBytes = 0;
 
                 if (headersCount > 0)
                 {
@@ -1689,6 +1692,7 @@ private void doEncodeRecordCont(
 
                     encodeSlotBuffer.putBytes(encodeSlotLimit,  value.buffer(), value.offset(), length);
                     encodeSlotLimit += length;
+                    encodeableRecordValueBytes += length;
 
                     if ((flags & FLAGS_FIN) == 0)
                     {
@@ -1893,7 +1897,8 @@ private void doEncodeProduceRequest(
 
                 final ByteBuffer encodeSlotByteBuffer = encodePool.byteBuffer(encodeSlot);
                 final int encodeSlotBytePosition = encodeSlotByteBuffer.position();
-                encodeSlotByteBuffer.limit(encodeSlotBytePosition + encodeSlotLimit);
+                final int partialValueSize = flushFlags != FLAGS_FIN ? encodeableRecordValueBytes : 0;
+                encodeSlotByteBuffer.limit(encodeSlotBytePosition + encodeSlotLimit - partialValueSize);
                 encodeSlotByteBuffer.position(encodeSlotBytePosition + encodeSlotOffset + crcLimit);
 
                 final CRC32C crc = crc32c;

runtime/binding-kafka/src/test/java/io/aklivity/zilla/runtime/binding/kafka/internal/stream/ClientMergedIT.java

@@ -49,7 +49,7 @@ public class ClientMergedIT
         .countersBufferCapacity(8192)
         .configure(ENGINE_BUFFER_SLOT_CAPACITY, 8192)
         .configure(KAFKA_CLIENT_META_MAX_AGE_MILLIS, 1000)
-        .configure(KAFKA_CLIENT_PRODUCE_MAX_BYTES, 116)
+        .configure(KAFKA_CLIENT_PRODUCE_MAX_BYTES, 528)
         .configurationRoot("io/aklivity/zilla/specs/binding/kafka/config")
         .external("net0")
         .clean();
@@ -234,7 +234,7 @@ public void shouldProduceMergedMessageValues() throws Exception
     @Configure(
         name = "zilla.binding.kafka.client.produce.max.bytes",
         value = "200000")
-    @ScriptProperty("padding ${512 + 100}")
+    @ScriptProperty("padding ${512 + 512}")
     public void shouldProduceMergedMessageValue10k() throws Exception
     {
         k3po.finish();
@@ -248,7 +248,7 @@ public void shouldProduceMergedMessageValue10k() throws Exception
     @Configure(
         name = "zilla.binding.kafka.client.produce.max.bytes",
         value = "200000")
-    @ScriptProperty("padding ${512 + 100}")
+    @ScriptProperty("padding ${512 + 512}")
     public void shouldProduceMergedMessageValue100k() throws Exception
     {
         k3po.finish();

specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/produce/message.value.100k/client.rpt

@@ -73,7 +73,7 @@ read zilla:begin.ext ${kafka:beginEx()
 write zilla:data.ext ${kafka:dataEx()
                               .typeId(zilla:id("kafka"))
                               .produce()
-                                  .deferred(102400 - 8192 + 512 + 100)
+                                  .deferred(102400 - 8192 + 512 + 512)
                                   .timestamp(newTimestamp)
                                   .build()
                               .build()}

specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/produce/message.value.100k/server.rpt

@@ -18,7 +18,7 @@ property serverAddress "zilla://streams/app0"
 
 accept ${serverAddress}
     option zilla:window 8192
-    option zilla:padding 612
+    option zilla:padding 1024
     option zilla:transmission "half-duplex"
 
 accepted
@@ -71,7 +71,7 @@ write zilla:begin.ext ${kafka:beginEx()
 read zilla:data.ext ${kafka:matchDataEx()
                              .typeId(zilla:id("kafka"))
                              .produce()
-                                 .deferred(102400 - 8192 + 512 + 100)
+                                 .deferred(102400 - 8192 + 512 + 512)
                                  .build()
                              .build()}
 

specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/produce/message.value.10k/client.rpt

@@ -73,7 +73,7 @@ read zilla:begin.ext ${kafka:beginEx()
 write zilla:data.ext ${kafka:dataEx()
                               .typeId(zilla:id("kafka"))
                               .produce()
-                                  .deferred(10240 - 8192 + 512 + 100)
+                                  .deferred(10240 - 8192 + 512 + 512)
                                   .timestamp(newTimestamp)
                                   .build()
                               .build()}

specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/produce/message.value.10k/server.rpt

@@ -18,7 +18,7 @@ property serverAddress "zilla://streams/app0"
 
 accept ${serverAddress}
     option zilla:window 8192
-    option zilla:padding 612
+    option zilla:padding 1024
     option zilla:transmission "half-duplex"
 
 accepted
@@ -71,7 +71,7 @@ write zilla:begin.ext ${kafka:beginEx()
 read zilla:data.ext ${kafka:matchDataEx()
                              .typeId(zilla:id("kafka"))
                              .produce()
-                                 .deferred(10240 - 8192 + 512 + 100)
+                                 .deferred(10240 - 8192 + 512 + 512)
                                  .build()
                              .build()}
 read zilla:data.ext ${kafka:matchDataEx()

specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/produce/message.values.sequential/client.rpt

@@ -81,7 +81,7 @@ write zilla:data.ext ${kafka:dataEx()
                               .produce()
                                   .build()
                               .build()}
-write ${kafka:randomBytes(7580)}
+write ${kafka:randomBytes(8192-(512+512))}
 write flush
 
 write zilla:data.ext ${kafka:dataEx()

specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/application/produce/message.values.sequential/server.rpt

@@ -79,7 +79,7 @@ read zilla:data.ext ${kafka:matchDataEx()
                              .produce()
                                  .build()
                              .build()}
-read [0..7580]
+read [0..7168]
 
 read zilla:data.ext ${kafka:matchDataEx()
                              .typeId(zilla:id("kafka"))

specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/produce.v3/message.values.sequential/client.rpt

@@ -78,7 +78,7 @@ write zilla:begin.ext ${proxy:beginEx()
 
 connected
 
-write 7690
+write 7278
       0s
       3s
       ${newRequestId}
@@ -90,9 +90,9 @@ write 7690
       4s "test"
       1
       0
-      7650                               # record set size
+      7238                               # record set size
       0L                                 # first offset
-      7638                               # length
+      7226                               # length
       -1
       [0x02]
       0x4e8723aa
@@ -104,13 +104,13 @@ write 7690
       -1s
       -1
       1                                  # records
-      ${kafka:varint(7587)}
+      ${kafka:varint(7175)}
       [0x00]
       ${kafka:varint(0)}
       ${kafka:varint(0)}
       ${kafka:varint(-1)}                # key
-      ${kafka:varint(7580)}              # value
-      ${kafka:randomBytes(7580)}
+      ${kafka:varint(7168)}              # value
+      ${kafka:randomBytes(7168)}
       ${kafka:varint(0)}                 # headers
 
 read 44

specs/binding-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/kafka/streams/network/produce.v3/message.values.sequential/server.rpt

@@ -74,7 +74,7 @@ read zilla:begin.ext ${proxy:beginEx()
 
 connected
 
-read 7690
+read 7278
      0s
      3s
      (int:requestId)
@@ -86,9 +86,9 @@ read 7690
      4s "test"
      1
      0
-     7650                               # record set size
+     7238                               # record set size
      0L                                 # first offset
-     7638                               # length
+     7226                               # length
      -1
      [0x02]
      [0..4]
@@ -100,13 +100,13 @@ read 7690
      -1s
      -1
      1                                  # records
-     ${kafka:varint(7587)}
+     ${kafka:varint(7175)}
      [0x00]
      ${kafka:varint(0)}
      ${kafka:varint(0)}
      ${kafka:varint(-1)}                # key
-     ${kafka:varint(7580)}              # value
-     [0..7580]
+     ${kafka:varint(7168)}              # value
+     [0..7168]
      ${kafka:varint(0)}                 # headers
 
 write 44