UNDERTOW-1487: SimpleObjectPool.close may be invoked multiple times

core/src/main/java/io/undertow/conduits/DeflatingStreamSinkConduit.java

@@ -540,8 +540,8 @@ private void freeBuffer() {
             state = state & ~FLUSHING_BUFFER;
         }
         if (deflater != null) {
-            pooledObject.close();
             deflater = null;
+            pooledObject.close();
         }
     }
 }

core/src/main/java/io/undertow/util/SimpleObjectPool.java

@@ -50,13 +50,13 @@ public SimpleObjectPool(int poolSize, Supplier<T> supplier, Consumer<T> consumer
     }
 
     @Override
-    public PooledObject allocate() {
+    public PooledObject<T> allocate() {
         T obj = pool.poll();
         if(obj == null) {
             obj = supplier.get();
         }
         final T finObj = obj;
-        return new PooledObject() {
+        return new PooledObject<T>() {
 
             private volatile boolean closed = false;
 
@@ -70,10 +70,12 @@ public T getObject() {
 
             @Override
             public void close() {
-                closed = true;
-                recycler.accept(finObj);
-                if(!pool.offer(finObj)) {
-                    consumer.accept(finObj);
+                if (!closed) {
+                    closed = true;
+                    recycler.accept(finObj);
+                    if (!pool.offer(finObj)) {
+                        consumer.accept(finObj);
+                    }
                 }
             }
         };

core/src/test/java/io/undertow/util/SimpleObjectPoolTestCase.java

@@ -0,0 +1,53 @@
+package io.undertow.util;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.assertEquals;
+
+public class SimpleObjectPoolTestCase {
+
+    @Rule
+    public final ExpectedException expected = ExpectedException.none();
+
+    @Test
+    public void testObjectAlreadyReturned() {
+        SimpleObjectPool<Object> pool = new SimpleObjectPool<>(1, Object::new, obj -> {}, obj -> {});
+        PooledObject<Object> pooled = pool.allocate();
+        pooled.close();
+
+        expected.expect(IllegalStateException.class);
+        pooled.getObject();
+    }
+
+    @Test
+    public void testCloseMayBeInvokedMultipleTimesWhenObjectIsRecycled() {
+        AtomicInteger recycled = new AtomicInteger();
+        AtomicInteger destroyed = new AtomicInteger();
+        SimpleObjectPool<Object> pool = new SimpleObjectPool<>(
+                1, Object::new, obj -> recycled.incrementAndGet(), obj -> destroyed.incrementAndGet());
+        PooledObject<Object> pooled = pool.allocate();
+        pooled.close();
+        pooled.close();
+        assertEquals("Pooled object should only be recycled once", 1, recycled.get());
+        assertEquals("Pooled object should be queued for reuse, not destroyed", 0, destroyed.get());
+    }
+
+    @Test
+    public void testCloseMayBeInvokedMultipleTimesWhenObjectIsConsumed() {
+        AtomicInteger recycled = new AtomicInteger();
+        AtomicInteger destroyed = new AtomicInteger();
+        SimpleObjectPool<Object> pool = new SimpleObjectPool<>(
+                1, Object::new, obj -> recycled.incrementAndGet(), obj -> destroyed.incrementAndGet());
+        PooledObject<Object> initial = pool.allocate();
+        PooledObject<Object> pooled = pool.allocate();
+        initial.close(); // This object fills the queue so that 'pooled' should be destroyed
+        pooled.close();
+        pooled.close();
+        assertEquals("Each pooled object should be recycled", 2, recycled.get());
+        assertEquals("Pooled object should be destroyed exactly once", 1, destroyed.get());
+    }
+}