fix(grpc): handle NotImplementedError from add_done_callback in aio client

CHANGELOG.md

@@ -16,6 +16,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Bump `pylint` to `4.0.5`
   ([#4244](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4244))
 
+### Fixed
+
+- `opentelemetry-instrumentation-grpc`: Fix `NotImplementedError` when `grpc.aio` call object does not implement `add_done_callback` (e.g. when a user interceptor awaits the call)
+  ([#4429](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4429))
+
 ### Breaking changes
 
 - Drop Python 3.9 support

instrumentation/opentelemetry-instrumentation-grpc/src/opentelemetry/instrumentation/grpc/_aio_client.py

@@ -114,11 +114,15 @@ async def _wrap_unary_response(self, continuation, span):
             code = await call.code()
             details = await call.details()
 
-            call.add_done_callback(
-                _unary_done_callback(
-                    span, code, details, self._call_response_hook
-                )
-            )
+            callback = _unary_done_callback(span, code, details, self._call_response_hook)
+            try:
+                call.add_done_callback(callback)
+            except NotImplementedError:
+                # Some grpc.aio interceptors (e.g. interceptors that await the
+                # call object) wrap the call in a type that does not implement
+                # add_done_callback.  In that case call the callback immediately
+                # since code and details are already known.
+                callback(call)
 
             return call
         except grpc.aio.AioRpcError as exc:

instrumentation/opentelemetry-instrumentation-grpc/tests/test_aio_client_interceptor.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 from unittest import IsolatedAsyncioTestCase
+from unittest.mock import patch
 
 import grpc
 
@@ -340,3 +341,35 @@ async def test_stream_stream_with_suppress_key(self):
 
             spans = self.memory_exporter.get_finished_spans()
             self.assertEqual(len(spans), 0)
+
+    async def test_unary_unary_add_done_callback_not_implemented(self):
+        """Span should still be finished when add_done_callback raises NotImplementedError."""
+
+        original_method = grpc.aio.UnaryUnaryCall.add_done_callback
+
+        def _raise_not_implemented(self_call, callback):
+            raise NotImplementedError
+
+        with patch.object(
+            grpc.aio.UnaryUnaryCall,
+            "add_done_callback",
+            _raise_not_implemented,
+        ):
+            response = await simple_method(self._stub)
+            assert response.response_data == "data"
+
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 1)
+        span = spans[0]
+
+        self.assertEqual(span.name, "/GRPCTestServer/SimpleMethod")
+        self.assertIs(span.kind, trace.SpanKind.CLIENT)
+        self.assertSpanHasAttributes(
+            span,
+            {
+                RPC_METHOD: "SimpleMethod",
+                RPC_SERVICE: "GRPCTestServer",
+                RPC_SYSTEM: "grpc",
+                RPC_GRPC_STATUS_CODE: grpc.StatusCode.OK.value[0],
+            },
+        )