Temporary Kafka-rest change for v1 testing

kafka-rest/src/main/java/io/confluent/kafkarest/resources/AsyncResponses.java

@@ -37,6 +37,13 @@ public static <T> void asyncResume(AsyncResponse asyncResponse, CompletableFutur
     AsyncResponseBuilder.<T>from(Response.ok()).entity(entity).asyncResume(asyncResponse);
   }
 
+  public static <T> void asyncResume(
+      AsyncResponse asyncResponse, CompletableFuture<T> entity, boolean isDeleteAclCall) {
+    AsyncResponseBuilder.<T>from(Response.ok())
+        .entity(entity)
+        .asyncResume(asyncResponse, isDeleteAclCall);
+  }
+
   /** A analogous of {@link AsyncResponse} for {@link ResponseBuilder}. */
   public static final class AsyncResponseBuilder<T> {
 
@@ -124,5 +131,42 @@ public void asyncResume(AsyncResponse asyncResponse) {
             }
           });
     }
+
+    public void asyncResume(AsyncResponse asyncResponse, boolean isDeleteAclCall) {
+      if (entityFuture == null) {
+        throw new IllegalStateException();
+      }
+      entityFuture.whenComplete(
+          (entity, exception) -> {
+            if (exception == null) {
+              if (statusFunction != null) {
+                responseBuilder.status(statusFunction.apply(entity));
+              }
+              if (entityAnnotations != null) {
+                asyncResponse.resume(responseBuilder.entity(entity, entityAnnotations).build());
+              } else {
+                asyncResponse.resume(responseBuilder.entity(entity).build());
+              }
+            } else if (exception instanceof CompletionException) {
+              log.error(
+                  "Async response CompletionException with error response entity of type {}: {}",
+                  entity != null ? entity.getClass() : "unknown",
+                  entity,
+                  exception);
+              asyncResponse.resume(exception.getCause());
+            } else {
+              log.error(
+                  "Async response exception with error response entity of type {}: {}",
+                  entity != null ? entity.getClass() : "unknown",
+                  entity,
+                  exception);
+              asyncResponse.resume(exception);
+            }
+          });
+      if (isDeleteAclCall) {
+        log.error("Delete ACL call throws error");
+        throw new IllegalStateException("Response does not exist (likely recycled)");
+      }
+    }
   }
 }

kafka-rest/src/main/java/io/confluent/kafkarest/resources/v3/AclsResource.java

@@ -255,7 +255,7 @@ public void deleteAcls(
                                     .thenComparing(AclData::getPermission))
                             .collect(Collectors.toList())));
 
-    AsyncResponses.asyncResume(asyncResponse, response);
+    AsyncResponses.asyncResume(asyncResponse, response, true);
   }
 
   public AclData toAclData(Acl acl) {

pom.xml

@@ -7,7 +7,7 @@
     <parent>
         <groupId>io.confluent</groupId>
         <artifactId>rest-utils-parent</artifactId>
-        <version>[8.1.0-0, 8.1.1-0)</version>
+        <version>8.1.0-0</version>
     </parent>
 
     <artifactId>kafka-rest-parent</artifactId>