Ensure inflight state empty at end of test

If we push inflight state during the request and do not pop it on the
response (or erroneously push inflight state for a request that does
not have a response a la zero-ack produce), then we will leak memory.

Signed-off-by: Robert Young <[email protected]>

Author: robobario

kroxylicious-filters/kroxylicious-authorization/src/main/java/io/kroxylicious/filter/authorization/AuthorizationFilter.java

@@ -8,6 +8,7 @@
 package io.kroxylicious.filter.authorization;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.EnumMap;
 import java.util.HashMap;
 import java.util.List;
@@ -273,4 +274,10 @@ public void adjustResponse(ApiVersionsResponseData response) {
         apiVersions.removeAll(toRemove);
 
     }
+
+    @SuppressWarnings("java:S1452") // wildcard type expected
+    @VisibleForTesting
+    Map<Integer, InflightState<?>> inflightState() {
+        return Collections.unmodifiableMap(inflightState);
+    }
 }

kroxylicious-filters/kroxylicious-authorization/src/test/java/io/kroxylicious/filter/authorization/AuthorizationFilterTest.java

@@ -148,6 +148,11 @@ void authorization(ScenarioDefinition definition) {
         if (!mockUpstream.isFinished()) {
             throw new IllegalStateException("test has finished, but mock responses are still queued");
         }
+        // we expect that any inflight state pushed during a request is always popped on the corresponding response
+        // if it is non-empty then we may have a memory leak
+        assertThat(authorizationFilter.inflightState())
+                .describedAs("inflight state")
+                .isEmpty();
     }
 
     private static String toYaml(Object actualBody) {