core: handle removing partially-closed resources for throwing on close. Fixes #6002. (#6044)

core/src/main/java/io/grpc/internal/SharedResourceHolder.java

@@ -128,8 +128,14 @@ public final class SharedResourceHolder {
         // AppEngine must immediately release shared resources, particularly executors
         // which could retain request-scoped threads which become zombies after the request
         // completes.
-        resource.close(instance);
-        instances.remove(resource);
+        // We do not encourage exceptions to be thrown during close, but we would like it to
+        // be able to recover eventually and do not want future resource fetches reuse the broken
+        // one.
+        try {
+          resource.close(instance);
+        } finally {
+          instances.remove(resource);
+        }
       } else {
         Preconditions.checkState(cached.destroyTask == null, "Destroy task already scheduled");
         // Schedule a delayed task to destroy the resource.
@@ -142,11 +148,14 @@ public final class SharedResourceHolder {
             synchronized (SharedResourceHolder.this) {
               // Refcount may have gone up since the task was scheduled. Re-check it.
               if (cached.refcount == 0) {
-                resource.close(instance);
-                instances.remove(resource);
-                if (instances.isEmpty()) {
-                  destroyer.shutdown();
-                  destroyer = null;
+                try {
+                  resource.close(instance);
+                } finally {
+                  instances.remove(resource);
+                  if (instances.isEmpty()) {
+                    destroyer.shutdown();
+                    destroyer = null;
+                  }
                 }
               }
             }

core/src/test/java/io/grpc/internal/SharedResourceHolderTest.java

@@ -173,6 +173,34 @@ public class SharedResourceHolderTest {
     }
   }
 
+  @Test public void handleInstanceCloseError() {
+    class ExceptionOnCloseResource implements Resource<ResourceInstance> {
+      @Override
+      public ResourceInstance create() {
+        return new ResourceInstance();
+      }
+
+      @Override
+      public void close(ResourceInstance instance) {
+        throw new RuntimeException();
+      }
+    }
+
+    Resource<ResourceInstance> resource = new ExceptionOnCloseResource();
+    ResourceInstance instance = holder.getInternal(resource);
+    holder.releaseInternal(resource, instance);
+    MockScheduledFuture<?> scheduledDestroyTask = scheduledDestroyTasks.poll();
+    try {
+      scheduledDestroyTask.runTask();
+      fail("Should throw RuntimeException");
+    } catch (RuntimeException e) {
+      // expected
+    }
+
+    // Future resource fetches should not get the partially-closed one.
+    assertNotSame(instance, holder.getInternal(resource));
+  }
+
   private class MockExecutorFactory implements
       SharedResourceHolder.ScheduledExecutorFactory {
     @Override