Fix mqtt-kafka non compact test

aklivity · Jun 4, 2024 · 14fca76 · 14fca76
1 parent 1045888
commit 14fca76
Show file tree

Hide file tree

Showing 3 changed files with 23 additions and 2 deletions.
diff --git a/...io/aklivity/zilla/runtime/binding/mqtt/kafka/internal/stream/MqttKafkaSessionFactory.java b/...io/aklivity/zilla/runtime/binding/mqtt/kafka/internal/stream/MqttKafkaSessionFactory.java
@@ -3470,6 +3470,7 @@ protected void onKafkaBegin(
                         delegate.doMqttWindow(authorization, traceId, 0, 0, 0);
                         delegate.doMqttReset(traceId, mqttResetEx);
                         events.onMqttConnectionReset(traceId, routedId, MQTT_NON_COMPACT_SESSIONS_TOPIC);
+                        doKafkaWindow(traceId, authorization, 0, 0, 0, 0, 0);
                         doKafkaAbort(traceId, authorization);
                         break onKafkaBegin;
                     }
@@ -3630,6 +3631,20 @@ protected void doKafkaBegin(
                 delegate.sessionId, server, capabilities);
         }
 
+        private void doKafkaWindow(
+            long traceId,
+            long authorization,
+            long budgetId,
+            int capabilities,
+            long replySeq,
+            long replyAck,
+            int replyMax)
+        {
+
+            doWindow(kafka, originId, routedId, replyId, replySeq, replyAck, replyMax,
+                traceId, authorization, budgetId, replyPad, 0, capabilities);
+        }
+
         private void cancelWillSignal(
             long authorization,
             long traceId)

diff --git a/...s/binding/mqtt/kafka/streams/kafka/session.reject.non.compacted.sessions.topic/client.rpt b/...s/binding/mqtt/kafka/streams/kafka/session.reject.non.compacted.sessions.topic/client.rpt
@@ -110,4 +110,7 @@ read zilla:begin.ext ${kafka:beginEx()
                                 .build()
                             .build()}
 
-connected
+connected
+
+write abort
+read aborted
diff --git a/...s/binding/mqtt/kafka/streams/kafka/session.reject.non.compacted.sessions.topic/server.rpt b/...s/binding/mqtt/kafka/streams/kafka/session.reject.non.compacted.sessions.topic/server.rpt
@@ -105,4 +105,7 @@ write zilla:begin.ext ${kafka:beginEx()
                                 .build()
                             .build()}
 
-connect abort
+connected
+
+read aborted
+write abort