Checkpoint

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

@@ -643,28 +643,33 @@ private void onKafkaData(
             }
             else
             {
+                String8FW encodedId = null;
+                OctetsFW key = null;
                 final int flags = data.flags();
                 final OctetsFW payload = data.payload();
-                final OctetsFW extension = data.extension();
-                final ExtensionFW dataEx = extension.get(extensionRO::tryWrap);
-                final KafkaDataExFW kafkaDataEx =
-                        dataEx != null && dataEx.typeId() == kafkaTypeId ? extension.get(kafkaDataExRO::tryWrap) : null;
-                final KafkaMergedDataExFW kafkaMergedDataEx =
-                        kafkaDataEx != null && kafkaDataEx.kind() == KafkaDataExFW.KIND_MERGED ? kafkaDataEx.merged() : null;
-                final Array32FW<KafkaOffsetFW> progress = kafkaMergedDataEx != null ? kafkaMergedDataEx.progress() : null;
-                final OctetsFW key = kafkaMergedDataEx != null ? kafkaMergedDataEx.key().value() : null;
-                final Array32FW<KafkaHeaderFW> headers = kafkaMergedDataEx != null ? kafkaMergedDataEx.headers() : null;
-                final KafkaHeaderFW etag = headers.matchFirst(h -> HEADER_NAME_ETAG.value().equals(h.name().value()));
 
-                String8FW encodedId = null;
-                switch (delegate.resolved.eventId())
+                if ((flags & 0x02) != 0x00)
                 {
-                case EVENT_ID_KEY64_AND_ETAG:
-                    encodedId = sseEventId.encodeKeyAndProgress(key, progress, etag);
-                    break;
-                case EVENT_ID_ETAG_ONLY:
-                    encodedId = sseEventId.encodeProgressOnly(progress, etag);
-                    break;
+                    final OctetsFW extension = data.extension();
+                    final ExtensionFW dataEx = extension.get(extensionRO::tryWrap);
+                    final KafkaDataExFW kafkaDataEx =
+                        dataEx != null && dataEx.typeId() == kafkaTypeId ? extension.get(kafkaDataExRO::tryWrap) : null;
+                    final KafkaMergedDataExFW kafkaMergedDataEx =
+                        kafkaDataEx != null && kafkaDataEx.kind() == KafkaDataExFW.KIND_MERGED ? kafkaDataEx.merged() : null;
+                    final Array32FW<KafkaOffsetFW> progress = kafkaMergedDataEx != null ? kafkaMergedDataEx.progress() : null;
+                    key = kafkaMergedDataEx != null ? kafkaMergedDataEx.key().value() : null;
+                    final Array32FW<KafkaHeaderFW> headers = kafkaMergedDataEx != null ? kafkaMergedDataEx.headers() : null;
+                    final KafkaHeaderFW etag = headers.matchFirst(h -> HEADER_NAME_ETAG.value().equals(h.name().value()));
+
+                    switch (delegate.resolved.eventId())
+                    {
+                    case EVENT_ID_KEY64_AND_ETAG:
+                        encodedId = sseEventId.encodeKeyAndProgress(key, progress, etag);
+                        break;
+                    case EVENT_ID_ETAG_ONLY:
+                        encodedId = sseEventId.encodeProgressOnly(progress, etag);
+                        break;
+                    }
                 }
 
                 final String8FW eventType = payload == null ? EVENT_TYPE_DELETE : EVENT_TYPE_MESSAGE;

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

@@ -157,6 +157,16 @@ public void shouldReceiveServerSentMessagesWithEtag() throws Exception
         k3po.finish();
     }
 
+    @Test
+    @Configuration("proxy.with.topic.yaml")
+    @Specification({
+        "${sse}/server.sent.messages.with.null.etag/client",
+        "${kafka}/server.sent.messages.with.null.etag/server"})
+    public void shouldReceiveServerSentMessagesWithNullEtag() throws Exception
+    {
+        k3po.finish();
+    }
+
     @Test
     @Configuration("proxy.with.topic.and.event.id.yaml")
     @Specification({

specs/binding-kafka.spec/src/main/java/io/aklivity/zilla/specs/binding/kafka/internal/KafkaFunctions.java

@@ -397,6 +397,7 @@ public KafkaFilterBuilder<T> header(
             String name,
             String value)
         {
+
             if (value == null)
             {
                 nameRO.wrap(name.getBytes(UTF_8));
@@ -1338,6 +1339,11 @@ public KafkaMergedDataExBuilder header(
                 return this;
             }
 
+            public KafkaMergedDataExBuilder headersNull()
+            {
+                return this;
+            }
+
             public KafkaMergedDataExBuilder headerNull(
                 String name)
             {

specs/binding-sse-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/sse/kafka/streams/kafka/server.sent.messages.with.null.etag/client.rpt

@@ -0,0 +1,40 @@
+#
+# Copyright 2021-2023 Aklivity Inc
+#
+# Licensed under the Aklivity Community License (the "License"); you may not use
+# this file except in compliance with the License.  You may obtain a copy of the
+# License at
+#
+#   https://www.aklivity.io/aklivity-community-license/
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations under the License.
+#
+
+connect "zilla://streams/kafka0"
+    option zilla:window 8192
+    option zilla:transmission "duplex"
+
+write zilla:begin.ext ${kafka:beginEx()
+                             .typeId(zilla:id("kafka"))
+                             .merged()
+                                 .capabilities("FETCH_ONLY")
+                                 .topic("test")
+                                 .partition(-1, -2)
+                                 .build()
+                             .build()}
+
+connected
+
+read zilla:data.ext ${kafka:matchDataEx()
+                           .typeId(zilla:id("kafka"))
+                           .merged()
+                               .partition(0, 1, 2)
+                               .progress(0, 2)
+                               .progress(1, 1)
+                               .key("key")
+                               .build()
+                           .build()}
+read "Hello, world"

specs/binding-sse-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/sse/kafka/streams/kafka/server.sent.messages.with.null.etag/server.rpt

@@ -0,0 +1,46 @@
+#
+# Copyright 2021-2023 Aklivity Inc
+#
+# Licensed under the Aklivity Community License (the "License"); you may not use
+# this file except in compliance with the License.  You may obtain a copy of the
+# License at
+#
+#   https://www.aklivity.io/aklivity-community-license/
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations under the License.
+#
+
+accept "zilla://streams/kafka0"
+    option zilla:window 8192
+    option zilla:transmission "duplex"
+
+accepted
+
+read zilla:begin.ext ${kafka:beginEx()
+                            .typeId(zilla:id("kafka"))
+                            .merged()
+                                .capabilities("FETCH_ONLY")
+                                .topic("test")
+                                .partition(-1, -2)
+                                .build()
+                            .build()}
+
+connected
+
+write zilla:data.ext ${kafka:dataEx()
+                            .typeId(zilla:id("kafka"))
+                            .merged()
+                                .timestamp(kafka:timestamp())
+                                .partition(0, 1, 2)
+                                .progress(0, 2)
+                                .progress(1, 1)
+                                .key("key")
+                                .headersNull()
+                                .build()
+                            .build()}
+write "Hello, world"
+write flush
+

specs/binding-sse-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/sse/kafka/streams/sse/server.sent.messages.with.null.etag/client.rpt

@@ -0,0 +1,35 @@
+#
+# Copyright 2021-2023 Aklivity Inc
+#
+# Licensed under the Aklivity Community License (the "License"); you may not use
+# this file except in compliance with the License.  You may obtain a copy of the
+# License at
+#
+#   https://www.aklivity.io/aklivity-community-license/
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations under the License.
+#
+
+connect "zilla://streams/sse0"
+        option zilla:window 8192
+        option zilla:transmission "duplex"
+
+write zilla:begin.ext ${sse:beginEx()
+                           .typeId(zilla:id("sse"))
+                           .scheme("http")
+                           .authority("localhost:8080")
+                           .path("/test")
+                           .lastId(null)
+                           .build()}
+
+connected
+
+read zilla:data.ext ${sse:matchDataEx()
+                         .typeId(zilla:id("sse"))
+                         .id(null)
+                         .type(null)
+                         .build()}
+read "Hello, world"

specs/binding-sse-kafka.spec/src/main/scripts/io/aklivity/zilla/specs/binding/sse/kafka/streams/sse/server.sent.messages.with.null.etag/server.rpt

@@ -0,0 +1,38 @@
+#
+# Copyright 2021-2023 Aklivity Inc
+#
+# Licensed under the Aklivity Community License (the "License"); you may not use
+# this file except in compliance with the License.  You may obtain a copy of the
+# License at
+#
+#   https://www.aklivity.io/aklivity-community-license/
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations under the License.
+#
+
+accept "zilla://streams/sse0"
+       option zilla:window 8192
+       option zilla:transmission "duplex"
+accepted
+
+read zilla:begin.ext ${sse:beginEx()
+                          .typeId(zilla:id("sse"))
+                          .scheme("http")
+                          .authority("localhost:8080")
+                          .path("/test")
+                          .lastId(null)
+                          .build()}
+
+connected
+
+write flush
+
+write zilla:data.ext ${sse:dataEx()
+                          .typeId(zilla:id("sse"))
+                          .id("AQQABAIC/revision=42")
+                          .build()}
+write "Hello, world"
+