diff --git a/BUILD.bazel b/BUILD.bazel
index 38ae745d0d..b062e9ca78 100644
--- a/BUILD.bazel
+++ b/BUILD.bazel
@@ -51,7 +51,7 @@ toolchain(
 
 py_binary(
     name = "gapic_plugin",
-    srcs = glob(["gapic/**/*.py"]),
+    srcs = glob(["gapic/**/*.py", "google/**/*.py"]),
     data = [":pandoc_binary"] + glob([
         "gapic/**/*.j2",
         "gapic/**/.*.j2",
diff --git a/gapic/schema/wrappers.py b/gapic/schema/wrappers.py
index cbefe83c2b..eef7b9f14c 100644
--- a/gapic/schema/wrappers.py
+++ b/gapic/schema/wrappers.py
@@ -41,6 +41,7 @@
 from google.api import resource_pb2
 from google.api_core import exceptions
 from google.api_core import path_template
+from google.cloud import extended_operations_pb2 as ex_ops_pb2
 from google.protobuf import descriptor_pb2  # type: ignore
 from google.protobuf.json_format import MessageToDict  # type: ignore
 
@@ -344,6 +345,39 @@ def oneof_fields(self, include_optional=False):
 
         return oneof_fields
 
+    @utils.cached_property
+    def is_diregapic_operation(self) -> bool:
+        if not self.name == "Operation":
+            return False
+
+        name, status, error_code, error_message = False, False, False, False
+        duplicate_msg = f"Message '{self.name}' has multiple fields with the same operation response mapping: {{}}"
+        for f in self.field:
+            maybe_op_mapping = f.options.Extensions[ex_ops_pb2.operation_field]
+            OperationResponseMapping = ex_ops_pb2.OperationResponseMapping
+
+            if maybe_op_mapping == OperationResponseMapping.NAME:
+                if name:
+                    raise TypeError(duplicate_msg.format("name"))
+                name = True
+
+            if maybe_op_mapping == OperationResponseMapping.STATUS:
+                if status:
+                    raise TypeError(duplicate_msg.format("status"))
+                status = True
+
+            if maybe_op_mapping == OperationResponseMapping.ERROR_CODE:
+                if error_code:
+                    raise TypeError(duplicate_msg.format("error_code"))
+                error_code = True
+
+            if maybe_op_mapping == OperationResponseMapping.ERROR_MESSAGE:
+                if error_message:
+                    raise TypeError(duplicate_msg.format("error_message"))
+                error_message = True
+
+        return name and status and error_code and error_message
+
     @utils.cached_property
     def required_fields(self) -> Sequence['Field']:
         required_fields = [
@@ -765,6 +799,10 @@ class Method:
     def __getattr__(self, name):
         return getattr(self.method_pb, name)
 
+    @property
+    def is_operation_polling_method(self):
+        return self.output.is_diregapic_operation and self.options.Extensions[ex_ops_pb2.operation_polling_method]
+
     @utils.cached_property
     def client_output(self):
         return self._client_output(enable_asyncio=False)
@@ -838,6 +876,10 @@ def _client_output(self, enable_asyncio: bool):
         # Return the usual output.
         return self.output
 
+    @property
+    def operation_service(self) -> Optional[str]:
+        return self.options.Extensions[ex_ops_pb2.operation_service]
+
     @property
     def is_deprecated(self) -> bool:
         """Returns true if the method is deprecated, false otherwise."""
@@ -1172,6 +1214,10 @@ class Service:
     def __getattr__(self, name):
         return getattr(self.service_pb, name)
 
+    @property
+    def custom_polling_method(self) -> Optional[Method]:
+        return next((m for m in self.methods.values() if m.is_operation_polling_method), None)
+
     @property
     def client_name(self) -> str:
         """Returns the name of the generated client class"""
diff --git a/gapic/templates/%namespace/%name_%version/%sub/services/%service/async_client.py.j2 b/gapic/templates/%namespace/%name_%version/%sub/services/%service/async_client.py.j2
index bbdffc3ba8..2f30d6cfab 100644
--- a/gapic/templates/%namespace/%name_%version/%sub/services/%service/async_client.py.j2
+++ b/gapic/templates/%namespace/%name_%version/%sub/services/%service/async_client.py.j2
@@ -150,7 +150,9 @@ class {{ service.async_client_name }}:
         )
 
     {% for method in service.methods.values() %}
-    {%+ if not method.server_streaming %}async {% endif %}def {{ method.name|snake_case }}(self,
+      {% with method_name = method.name|snake_case + "_unary" if method.operation_service else method.name|snake_case %}
+    {%+ if not method.server_streaming %}async {% endif %}def {{ method_name }}(self,
+      {% endwith %}
             {% if not method.client_streaming %}
             request: Union[{{ method.input.ident }}, dict] = None,
             *,
diff --git a/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2 b/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2
index 917fbd0fe6..25e4f2ca54 100644
--- a/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2
+++ b/gapic/templates/%namespace/%name_%version/%sub/services/%service/client.py.j2
@@ -315,7 +315,11 @@ class {{ service.client_name }}(metaclass={{ service.client_name }}Meta):
 
 
     {% for method in service.methods.values() %}
+    {% if method.operation_service %}{# DIREGAPIC LRO #}
+    def {{ method.name|snake_case }}_unary(self,
+    {% else %}
     def {{ method.name|snake_case }}(self,
+    {% endif %}{# DIREGAPIC LRO #}
             {% if not method.client_streaming %}
             request: Union[{{ method.input.ident }}, dict] = None,
             *,
diff --git a/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2 b/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
index 541ff03124..b53afebe8d 100644
--- a/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
+++ b/gapic/templates/tests/unit/gapic/%name_%version/%sub/test_%service.py.j2
@@ -430,8 +430,8 @@ def test_{{ service.client_name|snake_case }}_client_options_from_dict():
 {% endif %}
 
 
-{% for method in service.methods.values() if 'grpc' in opts.transport %}
-def test_{{ method.name|snake_case }}(transport: str = 'grpc', request_type={{ method.input.ident }}):
+{% for method in service.methods.values() if 'grpc' in opts.transport %}{% with method_name = method.name|snake_case + "_unary" if method.operation_service else method.name|snake_case %}
+def test_{{ method_name }}(transport: str = 'grpc', request_type={{ method.input.ident }}):
     client = {{ service.client_name }}(
         credentials=ga_credentials.AnonymousCredentials(),
         transport=transport,
@@ -472,7 +472,7 @@ def test_{{ method.name|snake_case }}(transport: str = 'grpc', request_type={{ m
         {% if method.client_streaming %}
         response = client.{{ method.name|snake_case }}(iter(requests))
         {% else %}
-        response = client.{{ method.name|snake_case }}(request)
+        response = client.{{ method_name }}(request)
         {% endif %}
 
         # Establish that the underlying gRPC stub method was called.
@@ -512,12 +512,12 @@ def test_{{ method.name|snake_case }}(transport: str = 'grpc', request_type={{ m
     {% endif %}
 
 
-def test_{{ method.name|snake_case }}_from_dict():
-    test_{{ method.name|snake_case }}(request_type=dict)
+def test_{{ method_name }}_from_dict():
+    test_{{ method_name }}(request_type=dict)
 
 
 {% if not method.client_streaming %}
-def test_{{ method.name|snake_case }}_empty_call():
+def test_{{ method_name }}_empty_call():
     # This test is a coverage failsafe to make sure that totally empty calls,
     # i.e. request == None and no flattened fields passed, work.
     client = {{ service.client_name }}(
@@ -529,7 +529,7 @@ def test_{{ method.name|snake_case }}_empty_call():
     with mock.patch.object(
             type(client.transport.{{ method.name|snake_case }}),
             '__call__') as call:
-        client.{{ method.name|snake_case }}()
+        client.{{ method_name }}()
         call.assert_called()
         _, args, _ = call.mock_calls[0]
         {% if method.client_streaming %}
@@ -541,7 +541,7 @@ def test_{{ method.name|snake_case }}_empty_call():
 
 
 @pytest.mark.asyncio
-async def test_{{ method.name|snake_case }}_async(transport: str = 'grpc_asyncio', request_type={{ method.input.ident }}):
+async def test_{{ method_name }}_async(transport: str = 'grpc_asyncio', request_type={{ method.input.ident }}):
     client = {{ service.async_client_name }}(
         credentials=ga_credentials.AnonymousCredentials(),
         transport=transport,
@@ -589,7 +589,7 @@ async def test_{{ method.name|snake_case }}_async(transport: str = 'grpc_asyncio
         {% elif method.client_streaming and not method.server_streaming %}
         response = await (await client.{{ method.name|snake_case }}(iter(requests)))
         {% else %}
-        response = await client.{{ method.name|snake_case }}(request)
+        response = await client.{{ method_name }}(request)
         {% endif %}
 
         # Establish that the underlying gRPC stub method was called.
@@ -626,12 +626,12 @@ async def test_{{ method.name|snake_case }}_async(transport: str = 'grpc_asyncio
 
 
 @pytest.mark.asyncio
-async def test_{{ method.name|snake_case }}_async_from_dict():
-    await test_{{ method.name|snake_case }}_async(request_type=dict)
+async def test_{{ method_name }}_async_from_dict():
+    await test_{{ method_name }}_async(request_type=dict)
 
 
 {% if method.field_headers and not method.client_streaming %}
-def test_{{ method.name|snake_case }}_field_headers():
+def test_{{ method_name }}_field_headers():
     client = {{ service.client_name }}(
         credentials=ga_credentials.AnonymousCredentials(),
     )
@@ -657,7 +657,7 @@ def test_{{ method.name|snake_case }}_field_headers():
         {% else %}
         call.return_value = {{ method.output.ident }}()
         {% endif %}
-        client.{{ method.name|snake_case }}(request)
+        client.{{ method_name }}(request)
 
         # Establish that the underlying gRPC stub method was called.
         assert len(call.mock_calls) == 1
@@ -676,7 +676,7 @@ def test_{{ method.name|snake_case }}_field_headers():
 
 
 @pytest.mark.asyncio
-async def test_{{ method.name|snake_case }}_field_headers_async():
+async def test_{{ method_name }}_field_headers_async():
     client = {{ service.async_client_name }}(
         credentials=ga_credentials.AnonymousCredentials(),
     )
@@ -703,7 +703,7 @@ async def test_{{ method.name|snake_case }}_field_headers_async():
         {% else %}
         call.return_value = grpc_helpers_async.FakeUnaryUnaryCall({{ method.output.ident }}())
         {% endif %}
-        await client.{{ method.name|snake_case }}(request)
+        await client.{{ method_name }}(request)
 
         # Establish that the underlying gRPC stub method was called.
         assert len(call.mock_calls)
@@ -722,7 +722,7 @@ async def test_{{ method.name|snake_case }}_field_headers_async():
 {% endif %}
 
 {% if method.ident.package != method.input.ident.package %}
-def test_{{ method.name|snake_case }}_from_dict_foreign():
+def test_{{ method_name }}_from_dict_foreign():
     client = {{ service.client_name }}(
         credentials=ga_credentials.AnonymousCredentials(),
     )
@@ -740,7 +740,7 @@ def test_{{ method.name|snake_case }}_from_dict_foreign():
         {% else %}
         call.return_value = {{ method.output.ident }}()
         {% endif %}
-        response = client.{{ method.name|snake_case }}(request={
+        response = client.{{ method_name }}(request={
             {% for field in method.input.fields.values() %}
             '{{ field.name }}': {{ field.mock_value }},
             {% endfor %}
@@ -751,7 +751,7 @@ def test_{{ method.name|snake_case }}_from_dict_foreign():
 {% endif %}
 
 {% if method.flattened_fields %}
-def test_{{ method.name|snake_case }}_flattened():
+def test_{{ method_name }}_flattened():
     client = {{ service.client_name }}(
         credentials=ga_credentials.AnonymousCredentials(),
     )
@@ -772,7 +772,7 @@ def test_{{ method.name|snake_case }}_flattened():
         {% endif %}
         # Call the method with a truthy value for each flattened field,
         # using the keyword arguments to the method.
-        client.{{ method.name|snake_case }}(
+        client.{{ method_name }}(
             {% for field in method.flattened_fields.values() %}
             {{ field.name }}={{ field.mock_value }},
             {% endfor %}
@@ -807,7 +807,7 @@ def test_{{ method.name|snake_case }}_flattened():
 
 
 
-def test_{{ method.name|snake_case }}_flattened_error():
+def test_{{ method_name }}_flattened_error():
     client = {{ service.client_name }}(
         credentials=ga_credentials.AnonymousCredentials(),
     )
@@ -815,7 +815,7 @@ def test_{{ method.name|snake_case }}_flattened_error():
     # Attempting to call a method with both a request object and flattened
     # fields is an error.
     with pytest.raises(ValueError):
-        client.{{ method.name|snake_case }}(
+        client.{{ method_name }}(
             {{ method.input.ident }}(),
             {% for field in method.flattened_fields.values() %}
             {{ field.name }}={{ field.mock_value }},
@@ -824,7 +824,7 @@ def test_{{ method.name|snake_case }}_flattened_error():
 
 
 @pytest.mark.asyncio
-async def test_{{ method.name|snake_case }}_flattened_async():
+async def test_{{ method_name }}_flattened_async():
     client = {{ service.async_client_name }}(
         credentials=ga_credentials.AnonymousCredentials(),
     )
@@ -865,7 +865,7 @@ async def test_{{ method.name|snake_case }}_flattened_async():
         {% endif %}
         # Call the method with a truthy value for each flattened field,
         # using the keyword arguments to the method.
-        response = await client.{{ method.name|snake_case }}(
+        response = await client.{{ method_name }}(
             {% for field in method.flattened_fields.values() %}
             {{ field.name }}={{ field.mock_value }},
             {% endfor %}
@@ -900,7 +900,7 @@ async def test_{{ method.name|snake_case }}_flattened_async():
 
 
 @pytest.mark.asyncio
-async def test_{{ method.name|snake_case }}_flattened_error_async():
+async def test_{{ method_name }}_flattened_error_async():
     client = {{ service.async_client_name }}(
         credentials=ga_credentials.AnonymousCredentials(),
     )
@@ -908,7 +908,7 @@ async def test_{{ method.name|snake_case }}_flattened_error_async():
     # Attempting to call a method with both a request object and flattened
     # fields is an error.
     with pytest.raises(ValueError):
-        await client.{{ method.name|snake_case }}(
+        await client.{{ method_name }}(
             {{ method.input.ident }}(),
             {% for field in method.flattened_fields.values() %}
             {{ field.name }}={{ field.mock_value }},
@@ -918,7 +918,7 @@ async def test_{{ method.name|snake_case }}_flattened_error_async():
 
 
 {% if method.paged_result_field %}
-def test_{{ method.name|snake_case }}_pager():
+def test_{{ method_name }}_pager():
     client = {{ service.client_name }}(
         credentials=ga_credentials.AnonymousCredentials,
     )
@@ -968,7 +968,7 @@ def test_{{ method.name|snake_case }}_pager():
             )),
         )
         {% endif %}
-        pager = client.{{ method.name|snake_case }}(request={})
+        pager = client.{{ method_name }}(request={})
 
         assert pager._metadata == metadata
 
@@ -977,7 +977,7 @@ def test_{{ method.name|snake_case }}_pager():
         assert all(isinstance(i, {{ method.paged_result_field.type.ident }})
                    for i in results)
 
-def test_{{ method.name|snake_case }}_pages():
+def test_{{ method_name }}_pages():
     client = {{ service.client_name }}(
         credentials=ga_credentials.AnonymousCredentials,
     )
@@ -1014,12 +1014,12 @@ def test_{{ method.name|snake_case }}_pages():
             ),
             RuntimeError,
         )
-        pages = list(client.{{ method.name|snake_case }}(request={}).pages)
+        pages = list(client.{{ method_name }}(request={}).pages)
         for page_, token in zip(pages, ['abc','def','ghi', '']):
             assert page_.raw_page.next_page_token == token
 
 @pytest.mark.asyncio
-async def test_{{ method.name|snake_case }}_async_pager():
+async def test_{{ method_name }}_async_pager():
     client = {{ service.async_client_name }}(
         credentials=ga_credentials.AnonymousCredentials,
     )
@@ -1056,7 +1056,7 @@ async def test_{{ method.name|snake_case }}_async_pager():
             ),
             RuntimeError,
         )
-        async_pager = await client.{{ method.name|snake_case }}(request={},)
+        async_pager = await client.{{ method_name }}(request={},)
         assert async_pager.next_page_token == 'abc'
         responses = []
         async for response in async_pager:
@@ -1067,7 +1067,7 @@ async def test_{{ method.name|snake_case }}_async_pager():
                    for i in responses)
 
 @pytest.mark.asyncio
-async def test_{{ method.name|snake_case }}_async_pages():
+async def test_{{ method_name }}_async_pages():
     client = {{ service.async_client_name }}(
         credentials=ga_credentials.AnonymousCredentials,
     )
@@ -1105,23 +1105,23 @@ async def test_{{ method.name|snake_case }}_async_pages():
             RuntimeError,
         )
         pages = []
-        async for page_ in (await client.{{ method.name|snake_case }}(request={})).pages:
+        async for page_ in (await client.{{ method_name }}(request={})).pages:
             pages.append(page_)
         for page_, token in zip(pages, ['abc','def','ghi', '']):
             assert page_.raw_page.next_page_token == token
 {% elif method.lro and "next_page_token" in method.lro.response_type.fields.keys() %}
-def test_{{ method.name|snake_case }}_raw_page_lro():
+def test_{{ method_name }}_raw_page_lro():
     response = {{ method.lro.response_type.ident }}()
     assert response.raw_page is response
-{% endif %} {# method.paged_result_field #}
+{% endif %}{# method.paged_result_field #}{% endwith %}{# method_name #}
 
 {% endfor %} {# method in methods for grpc #}
 
 {% for method in service.methods.values() if 'rest' in opts.transport and
-    method.http_options %}
+    method.http_options %}{% with method_name = method.name|snake_case + "_unary" if method.operation_service else method.name|snake_case %}
 {# TODO(kbandes): remove this if condition when lro and streaming are supported. #}
 {% if not method.lro and not (method.server_streaming or method.client_streaming) %}
-def test_{{ method.name|snake_case }}_rest(transport: str = 'rest', request_type={{ method.input.ident }}):
+def test_{{ method_name }}_rest(transport: str = 'rest', request_type={{ method.input.ident }}):
     client = {{ service.client_name }}(
         credentials=ga_credentials.AnonymousCredentials(),
         transport=transport,
@@ -1177,7 +1177,7 @@ def test_{{ method.name|snake_case }}_rest(transport: str = 'rest', request_type
         {% if method.client_streaming %}
         response = client.{{ method.name|snake_case }}(iter(requests))
         {% else %}
-        response = client.{{ method.name|snake_case }}(request)
+        response = client.{{ method_name }}(request)
         {% endif %}
 
     {% if "next_page_token" in method.output.fields.values()|map(attribute='name') and not method.paged_result_field %}
@@ -1204,7 +1204,7 @@ def test_{{ method.name|snake_case }}_rest(transport: str = 'rest', request_type
     {% endif %}
 
 
-def test_{{ method.name|snake_case }}_rest_bad_request(transport: str = 'rest', request_type={{ method.input.ident }}):
+def test_{{ method_name }}_rest_bad_request(transport: str = 'rest', request_type={{ method.input.ident }}):
     client = {{ service.client_name }}(
         credentials=ga_credentials.AnonymousCredentials(),
         transport=transport,
@@ -1233,16 +1233,16 @@ def test_{{ method.name|snake_case }}_rest_bad_request(transport: str = 'rest',
         {% if method.client_streaming %}
         client.{{ method.name|snake_case }}(iter(requests))
         {% else %}
-        client.{{ method.name|snake_case }}(request)
+        client.{{ method_name }}(request)
         {% endif %}
 
 
-def test_{{ method.name|snake_case }}_rest_from_dict():
-    test_{{ method.name|snake_case }}_rest(request_type=dict)
+def test_{{ method_name }}_rest_from_dict():
+    test_{{ method_name }}_rest(request_type=dict)
 
 
 {% if method.flattened_fields %}
-def test_{{ method.name|snake_case }}_rest_flattened(transport: str = 'rest'):
+def test_{{ method_name }}_rest_flattened(transport: str = 'rest'):
     client = {{ service.client_name }}(
         credentials=ga_credentials.AnonymousCredentials(),
         transport=transport,
@@ -1286,7 +1286,7 @@ def test_{{ method.name|snake_case }}_rest_flattened(transport: str = 'rest'):
             {% endfor %}
         )
         mock_args.update(sample_request)
-        client.{{ method.name|snake_case }}(**mock_args)
+        client.{{ method_name }}(**mock_args)
 
         # Establish that the underlying call was made with the expected
         # request object values.
@@ -1298,7 +1298,7 @@ def test_{{ method.name|snake_case }}_rest_flattened(transport: str = 'rest'):
         {# TODO(kbandes) - reverse-transcode request args to check all request fields #}
 
 
-def test_{{ method.name|snake_case }}_rest_flattened_error(transport: str = 'rest'):
+def test_{{ method_name }}_rest_flattened_error(transport: str = 'rest'):
     client = {{ service.client_name }}(
         credentials=ga_credentials.AnonymousCredentials(),
         transport=transport,
@@ -1307,7 +1307,7 @@ def test_{{ method.name|snake_case }}_rest_flattened_error(transport: str = 'res
     # Attempting to call a method with both a request object and flattened
     # fields is an error.
     with pytest.raises(ValueError):
-        client.{{ method.name|snake_case }}(
+        client.{{ method_name }}(
             {{ method.input.ident }}(),
             {% for field in method.flattened_fields.values() %}
             {{ field.name }}={{ field.mock_value }},
@@ -1317,7 +1317,7 @@ def test_{{ method.name|snake_case }}_rest_flattened_error(transport: str = 'res
 
 
 {% if method.paged_result_field %}
-def test_{{ method.name|snake_case }}_rest_pager():
+def test_{{ method_name }}_rest_pager():
     client = {{ service.client_name }}(
         credentials=ga_credentials.AnonymousCredentials(),
     )
@@ -1401,7 +1401,7 @@ def test_{{ method.name|snake_case }}_rest_pager():
             {% endif %}
             {% endfor %}
 
-            pager = client.{{ method.name|snake_case }}(request=sample_request)
+            pager = client.{{ method_name }}(request=sample_request)
 
             {% if method.paged_result_field.map %}
             assert isinstance(pager.get('a'), {{ method.paged_result_field.type.fields.get('value').ident }})
@@ -1425,7 +1425,7 @@ def test_{{ method.name|snake_case }}_rest_pager():
                     for i in results)
             {% endif %}
 
-            pages = list(client.{{ method.name|snake_case }}(request=sample_request).pages)
+            pages = list(client.{{ method_name }}(request=sample_request).pages)
             for page_, token in zip(pages, ['abc','def','ghi', '']):
                 assert page_.raw_page.next_page_token == token
 
@@ -1433,7 +1433,7 @@ def test_{{ method.name|snake_case }}_rest_pager():
 {% endif %} {# paged methods #}
 {%- else %}
 
-def test_{{ method.name|snake_case }}_rest_error():
+def test_{{ method_name }}_rest_error():
     client = {{ service.client_name }}(
         credentials=ga_credentials.AnonymousCredentials(),
         transport='rest'
@@ -1442,7 +1442,7 @@ def test_{{ method.name|snake_case }}_rest_error():
     # Since a `google.api.http` annotation is required for using a rest transport
     # method, this should error.
     with pytest.raises(RuntimeError) as runtime_error:
-        client.{{ method.name|snake_case }}({})
+        client.{{ method_name }}({})
     assert ('Cannot define a method without a valid `google.api.http` annotation.'
         in str(runtime_error.value))
     {%- else %}
@@ -1450,11 +1450,10 @@ def test_{{ method.name|snake_case }}_rest_error():
     # TODO(yon-mg): Remove when this method has a working implementation
     #               or testing straegy
     with pytest.raises(NotImplementedError):
-        client.{{ method.name|snake_case }}({})
+        client.{{ method_name }}({})
 
     {%- endif %}
-{% endif %}
-
+{% endif %}{% endwith %}{# method_name #}
 
 {% endfor -%} {#- method in methods for rest #}
 def test_credentials_transport_error():
diff --git a/google/cloud/extended_operations_pb2.py b/google/cloud/extended_operations_pb2.py
new file mode 100755
index 0000000000..439f7f0be3
--- /dev/null
+++ b/google/cloud/extended_operations_pb2.py
@@ -0,0 +1,132 @@
+# -*- coding: utf-8 -*-
+# Generated by the protocol buffer compiler.  DO NOT EDIT!
+# source: google/cloud/extended_operations.proto
+"""Generated protocol buffer code."""
+from google.protobuf.internal import enum_type_wrapper
+from google.protobuf import descriptor as _descriptor
+from google.protobuf import message as _message
+from google.protobuf import reflection as _reflection
+from google.protobuf import symbol_database as _symbol_database
+# @@protoc_insertion_point(imports)
+
+_sym_db = _symbol_database.Default()
+
+
+from google.protobuf import descriptor_pb2 as google_dot_protobuf_dot_descriptor__pb2
+
+
+DESCRIPTOR = _descriptor.FileDescriptor(
+  name='google/cloud/extended_operations.proto',
+  package='google.cloud',
+  syntax='proto3',
+  serialized_options=b'\n\020com.google.cloudB\027ExtendedOperationsProtoP\001ZCgoogle.golang.org/genproto/googleapis/cloud/extendedops;extendedops\242\002\004GAPI',
+  create_key=_descriptor._internal_create_key,
+  serialized_pb=b'\n&google/cloud/extended_operations.proto\x12\x0cgoogle.cloud\x1a google/protobuf/descriptor.proto*b\n\x18OperationResponseMapping\x12\r\n\tUNDEFINED\x10\x00\x12\x08\n\x04NAME\x10\x01\x12\n\n\x06STATUS\x10\x02\x12\x0e\n\nERROR_CODE\x10\x03\x12\x11\n\rERROR_MESSAGE\x10\x04:_\n\x0foperation_field\x12\x1d.google.protobuf.FieldOptions\x18\xfd\x08 \x01(\x0e\x32&.google.cloud.OperationResponseMapping:?\n\x17operation_request_field\x12\x1d.google.protobuf.FieldOptions\x18\xfe\x08 \x01(\t:@\n\x18operation_response_field\x12\x1d.google.protobuf.FieldOptions\x18\xff\x08 \x01(\t::\n\x11operation_service\x12\x1e.google.protobuf.MethodOptions\x18\xe1\t \x01(\t:A\n\x18operation_polling_method\x12\x1e.google.protobuf.MethodOptions\x18\xe2\t \x01(\x08\x42y\n\x10\x63om.google.cloudB\x17\x45xtendedOperationsProtoP\x01ZCgoogle.golang.org/genproto/googleapis/cloud/extendedops;extendedops\xa2\x02\x04GAPIb\x06proto3'
+  ,
+  dependencies=[google_dot_protobuf_dot_descriptor__pb2.DESCRIPTOR,])
+
+_OPERATIONRESPONSEMAPPING = _descriptor.EnumDescriptor(
+  name='OperationResponseMapping',
+  full_name='google.cloud.OperationResponseMapping',
+  filename=None,
+  file=DESCRIPTOR,
+  create_key=_descriptor._internal_create_key,
+  values=[
+    _descriptor.EnumValueDescriptor(
+      name='UNDEFINED', index=0, number=0,
+      serialized_options=None,
+      type=None,
+      create_key=_descriptor._internal_create_key),
+    _descriptor.EnumValueDescriptor(
+      name='NAME', index=1, number=1,
+      serialized_options=None,
+      type=None,
+      create_key=_descriptor._internal_create_key),
+    _descriptor.EnumValueDescriptor(
+      name='STATUS', index=2, number=2,
+      serialized_options=None,
+      type=None,
+      create_key=_descriptor._internal_create_key),
+    _descriptor.EnumValueDescriptor(
+      name='ERROR_CODE', index=3, number=3,
+      serialized_options=None,
+      type=None,
+      create_key=_descriptor._internal_create_key),
+    _descriptor.EnumValueDescriptor(
+      name='ERROR_MESSAGE', index=4, number=4,
+      serialized_options=None,
+      type=None,
+      create_key=_descriptor._internal_create_key),
+  ],
+  containing_type=None,
+  serialized_options=None,
+  serialized_start=90,
+  serialized_end=188,
+)
+_sym_db.RegisterEnumDescriptor(_OPERATIONRESPONSEMAPPING)
+
+OperationResponseMapping = enum_type_wrapper.EnumTypeWrapper(_OPERATIONRESPONSEMAPPING)
+UNDEFINED = 0
+NAME = 1
+STATUS = 2
+ERROR_CODE = 3
+ERROR_MESSAGE = 4
+
+OPERATION_FIELD_FIELD_NUMBER = 1149
+operation_field = _descriptor.FieldDescriptor(
+  name='operation_field', full_name='google.cloud.operation_field', index=0,
+  number=1149, type=14, cpp_type=8, label=1,
+  has_default_value=False, default_value=0,
+  message_type=None, enum_type=None, containing_type=None,
+  is_extension=True, extension_scope=None,
+  serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key)
+OPERATION_REQUEST_FIELD_FIELD_NUMBER = 1150
+operation_request_field = _descriptor.FieldDescriptor(
+  name='operation_request_field', full_name='google.cloud.operation_request_field', index=1,
+  number=1150, type=9, cpp_type=9, label=1,
+  has_default_value=False, default_value=b"".decode('utf-8'),
+  message_type=None, enum_type=None, containing_type=None,
+  is_extension=True, extension_scope=None,
+  serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key)
+OPERATION_RESPONSE_FIELD_FIELD_NUMBER = 1151
+operation_response_field = _descriptor.FieldDescriptor(
+  name='operation_response_field', full_name='google.cloud.operation_response_field', index=2,
+  number=1151, type=9, cpp_type=9, label=1,
+  has_default_value=False, default_value=b"".decode('utf-8'),
+  message_type=None, enum_type=None, containing_type=None,
+  is_extension=True, extension_scope=None,
+  serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key)
+OPERATION_SERVICE_FIELD_NUMBER = 1249
+operation_service = _descriptor.FieldDescriptor(
+  name='operation_service', full_name='google.cloud.operation_service', index=3,
+  number=1249, type=9, cpp_type=9, label=1,
+  has_default_value=False, default_value=b"".decode('utf-8'),
+  message_type=None, enum_type=None, containing_type=None,
+  is_extension=True, extension_scope=None,
+  serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key)
+OPERATION_POLLING_METHOD_FIELD_NUMBER = 1250
+operation_polling_method = _descriptor.FieldDescriptor(
+  name='operation_polling_method', full_name='google.cloud.operation_polling_method', index=4,
+  number=1250, type=8, cpp_type=7, label=1,
+  has_default_value=False, default_value=False,
+  message_type=None, enum_type=None, containing_type=None,
+  is_extension=True, extension_scope=None,
+  serialized_options=None, file=DESCRIPTOR,  create_key=_descriptor._internal_create_key)
+
+DESCRIPTOR.enum_types_by_name['OperationResponseMapping'] = _OPERATIONRESPONSEMAPPING
+DESCRIPTOR.extensions_by_name['operation_field'] = operation_field
+DESCRIPTOR.extensions_by_name['operation_request_field'] = operation_request_field
+DESCRIPTOR.extensions_by_name['operation_response_field'] = operation_response_field
+DESCRIPTOR.extensions_by_name['operation_service'] = operation_service
+DESCRIPTOR.extensions_by_name['operation_polling_method'] = operation_polling_method
+_sym_db.RegisterFileDescriptor(DESCRIPTOR)
+
+operation_field.enum_type = _OPERATIONRESPONSEMAPPING
+google_dot_protobuf_dot_descriptor__pb2.FieldOptions.RegisterExtension(operation_field)
+google_dot_protobuf_dot_descriptor__pb2.FieldOptions.RegisterExtension(operation_request_field)
+google_dot_protobuf_dot_descriptor__pb2.FieldOptions.RegisterExtension(operation_response_field)
+google_dot_protobuf_dot_descriptor__pb2.MethodOptions.RegisterExtension(operation_service)
+google_dot_protobuf_dot_descriptor__pb2.MethodOptions.RegisterExtension(operation_polling_method)
+
+DESCRIPTOR._options = None
+# @@protoc_insertion_point(module_scope)
diff --git a/tests/fragments/google/api/annotations.proto b/tests/fragments/google/api/annotations.proto
new file mode 100644
index 0000000000..efdab3db6c
--- /dev/null
+++ b/tests/fragments/google/api/annotations.proto
@@ -0,0 +1,31 @@
+// Copyright 2015 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+syntax = "proto3";
+
+package google.api;
+
+import "google/api/http.proto";
+import "google/protobuf/descriptor.proto";
+
+option go_package = "google.golang.org/genproto/googleapis/api/annotations;annotations";
+option java_multiple_files = true;
+option java_outer_classname = "AnnotationsProto";
+option java_package = "com.google.api";
+option objc_class_prefix = "GAPI";
+
+extend google.protobuf.MethodOptions {
+  // See `HttpRule`.
+  HttpRule http = 72295728;
+}
diff --git a/tests/fragments/google/api/http.proto b/tests/fragments/google/api/http.proto
new file mode 100644
index 0000000000..113fa936a0
--- /dev/null
+++ b/tests/fragments/google/api/http.proto
@@ -0,0 +1,375 @@
+// Copyright 2015 Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+syntax = "proto3";
+
+package google.api;
+
+option cc_enable_arenas = true;
+option go_package = "google.golang.org/genproto/googleapis/api/annotations;annotations";
+option java_multiple_files = true;
+option java_outer_classname = "HttpProto";
+option java_package = "com.google.api";
+option objc_class_prefix = "GAPI";
+
+// Defines the HTTP configuration for an API service. It contains a list of
+// [HttpRule][google.api.HttpRule], each specifying the mapping of an RPC method
+// to one or more HTTP REST API methods.
+message Http {
+  // A list of HTTP configuration rules that apply to individual API methods.
+  //
+  // **NOTE:** All service configuration rules follow "last one wins" order.
+  repeated HttpRule rules = 1;
+
+  // When set to true, URL path parameters will be fully URI-decoded except in
+  // cases of single segment matches in reserved expansion, where "%2F" will be
+  // left encoded.
+  //
+  // The default behavior is to not decode RFC 6570 reserved characters in multi
+  // segment matches.
+  bool fully_decode_reserved_expansion = 2;
+}
+
+// # gRPC Transcoding
+//
+// gRPC Transcoding is a feature for mapping between a gRPC method and one or
+// more HTTP REST endpoints. It allows developers to build a single API service
+// that supports both gRPC APIs and REST APIs. Many systems, including [Google
+// APIs](https://github.com/googleapis/googleapis),
+// [Cloud Endpoints](https://cloud.google.com/endpoints), [gRPC
+// Gateway](https://github.com/grpc-ecosystem/grpc-gateway),
+// and [Envoy](https://github.com/envoyproxy/envoy) proxy support this feature
+// and use it for large scale production services.
+//
+// `HttpRule` defines the schema of the gRPC/REST mapping. The mapping specifies
+// how different portions of the gRPC request message are mapped to the URL
+// path, URL query parameters, and HTTP request body. It also controls how the
+// gRPC response message is mapped to the HTTP response body. `HttpRule` is
+// typically specified as an `google.api.http` annotation on the gRPC method.
+//
+// Each mapping specifies a URL path template and an HTTP method. The path
+// template may refer to one or more fields in the gRPC request message, as long
+// as each field is a non-repeated field with a primitive (non-message) type.
+// The path template controls how fields of the request message are mapped to
+// the URL path.
+//
+// Example:
+//
+//     service Messaging {
+//       rpc GetMessage(GetMessageRequest) returns (Message) {
+//         option (google.api.http) = {
+//             get: "/v1/{name=messages/*}"
+//         };
+//       }
+//     }
+//     message GetMessageRequest {
+//       string name = 1; // Mapped to URL path.
+//     }
+//     message Message {
+//       string text = 1; // The resource content.
+//     }
+//
+// This enables an HTTP REST to gRPC mapping as below:
+//
+// HTTP | gRPC
+// -----|-----
+// `GET /v1/messages/123456`  | `GetMessage(name: "messages/123456")`
+//
+// Any fields in the request message which are not bound by the path template
+// automatically become HTTP query parameters if there is no HTTP request body.
+// For example:
+//
+//     service Messaging {
+//       rpc GetMessage(GetMessageRequest) returns (Message) {
+//         option (google.api.http) = {
+//             get:"/v1/messages/{message_id}"
+//         };
+//       }
+//     }
+//     message GetMessageRequest {
+//       message SubMessage {
+//         string subfield = 1;
+//       }
+//       string message_id = 1; // Mapped to URL path.
+//       int64 revision = 2;    // Mapped to URL query parameter `revision`.
+//       SubMessage sub = 3;    // Mapped to URL query parameter `sub.subfield`.
+//     }
+//
+// This enables a HTTP JSON to RPC mapping as below:
+//
+// HTTP | gRPC
+// -----|-----
+// `GET /v1/messages/123456?revision=2&sub.subfield=foo` |
+// `GetMessage(message_id: "123456" revision: 2 sub: SubMessage(subfield:
+// "foo"))`
+//
+// Note that fields which are mapped to URL query parameters must have a
+// primitive type or a repeated primitive type or a non-repeated message type.
+// In the case of a repeated type, the parameter can be repeated in the URL
+// as `...?param=A&param=B`. In the case of a message type, each field of the
+// message is mapped to a separate parameter, such as
+// `...?foo.a=A&foo.b=B&foo.c=C`.
+//
+// For HTTP methods that allow a request body, the `body` field
+// specifies the mapping. Consider a REST update method on the
+// message resource collection:
+//
+//     service Messaging {
+//       rpc UpdateMessage(UpdateMessageRequest) returns (Message) {
+//         option (google.api.http) = {
+//           patch: "/v1/messages/{message_id}"
+//           body: "message"
+//         };
+//       }
+//     }
+//     message UpdateMessageRequest {
+//       string message_id = 1; // mapped to the URL
+//       Message message = 2;   // mapped to the body
+//     }
+//
+// The following HTTP JSON to RPC mapping is enabled, where the
+// representation of the JSON in the request body is determined by
+// protos JSON encoding:
+//
+// HTTP | gRPC
+// -----|-----
+// `PATCH /v1/messages/123456 { "text": "Hi!" }` | `UpdateMessage(message_id:
+// "123456" message { text: "Hi!" })`
+//
+// The special name `*` can be used in the body mapping to define that
+// every field not bound by the path template should be mapped to the
+// request body.  This enables the following alternative definition of
+// the update method:
+//
+//     service Messaging {
+//       rpc UpdateMessage(Message) returns (Message) {
+//         option (google.api.http) = {
+//           patch: "/v1/messages/{message_id}"
+//           body: "*"
+//         };
+//       }
+//     }
+//     message Message {
+//       string message_id = 1;
+//       string text = 2;
+//     }
+//
+//
+// The following HTTP JSON to RPC mapping is enabled:
+//
+// HTTP | gRPC
+// -----|-----
+// `PATCH /v1/messages/123456 { "text": "Hi!" }` | `UpdateMessage(message_id:
+// "123456" text: "Hi!")`
+//
+// Note that when using `*` in the body mapping, it is not possible to
+// have HTTP parameters, as all fields not bound by the path end in
+// the body. This makes this option more rarely used in practice when
+// defining REST APIs. The common usage of `*` is in custom methods
+// which don't use the URL at all for transferring data.
+//
+// It is possible to define multiple HTTP methods for one RPC by using
+// the `additional_bindings` option. Example:
+//
+//     service Messaging {
+//       rpc GetMessage(GetMessageRequest) returns (Message) {
+//         option (google.api.http) = {
+//           get: "/v1/messages/{message_id}"
+//           additional_bindings {
+//             get: "/v1/users/{user_id}/messages/{message_id}"
+//           }
+//         };
+//       }
+//     }
+//     message GetMessageRequest {
+//       string message_id = 1;
+//       string user_id = 2;
+//     }
+//
+// This enables the following two alternative HTTP JSON to RPC mappings:
+//
+// HTTP | gRPC
+// -----|-----
+// `GET /v1/messages/123456` | `GetMessage(message_id: "123456")`
+// `GET /v1/users/me/messages/123456` | `GetMessage(user_id: "me" message_id:
+// "123456")`
+//
+// ## Rules for HTTP mapping
+//
+// 1. Leaf request fields (recursive expansion nested messages in the request
+//    message) are classified into three categories:
+//    - Fields referred by the path template. They are passed via the URL path.
+//    - Fields referred by the [HttpRule.body][google.api.HttpRule.body]. They are passed via the HTTP
+//      request body.
+//    - All other fields are passed via the URL query parameters, and the
+//      parameter name is the field path in the request message. A repeated
+//      field can be represented as multiple query parameters under the same
+//      name.
+//  2. If [HttpRule.body][google.api.HttpRule.body] is "*", there is no URL query parameter, all fields
+//     are passed via URL path and HTTP request body.
+//  3. If [HttpRule.body][google.api.HttpRule.body] is omitted, there is no HTTP request body, all
+//     fields are passed via URL path and URL query parameters.
+//
+// ### Path template syntax
+//
+//     Template = "/" Segments [ Verb ] ;
+//     Segments = Segment { "/" Segment } ;
+//     Segment  = "*" | "**" | LITERAL | Variable ;
+//     Variable = "{" FieldPath [ "=" Segments ] "}" ;
+//     FieldPath = IDENT { "." IDENT } ;
+//     Verb     = ":" LITERAL ;
+//
+// The syntax `*` matches a single URL path segment. The syntax `**` matches
+// zero or more URL path segments, which must be the last part of the URL path
+// except the `Verb`.
+//
+// The syntax `Variable` matches part of the URL path as specified by its
+// template. A variable template must not contain other variables. If a variable
+// matches a single path segment, its template may be omitted, e.g. `{var}`
+// is equivalent to `{var=*}`.
+//
+// The syntax `LITERAL` matches literal text in the URL path. If the `LITERAL`
+// contains any reserved character, such characters should be percent-encoded
+// before the matching.
+//
+// If a variable contains exactly one path segment, such as `"{var}"` or
+// `"{var=*}"`, when such a variable is expanded into a URL path on the client
+// side, all characters except `[-_.~0-9a-zA-Z]` are percent-encoded. The
+// server side does the reverse decoding. Such variables show up in the
+// [Discovery
+// Document](https://developers.google.com/discovery/v1/reference/apis) as
+// `{var}`.
+//
+// If a variable contains multiple path segments, such as `"{var=foo/*}"`
+// or `"{var=**}"`, when such a variable is expanded into a URL path on the
+// client side, all characters except `[-_.~/0-9a-zA-Z]` are percent-encoded.
+// The server side does the reverse decoding, except "%2F" and "%2f" are left
+// unchanged. Such variables show up in the
+// [Discovery
+// Document](https://developers.google.com/discovery/v1/reference/apis) as
+// `{+var}`.
+//
+// ## Using gRPC API Service Configuration
+//
+// gRPC API Service Configuration (service config) is a configuration language
+// for configuring a gRPC service to become a user-facing product. The
+// service config is simply the YAML representation of the `google.api.Service`
+// proto message.
+//
+// As an alternative to annotating your proto file, you can configure gRPC
+// transcoding in your service config YAML files. You do this by specifying a
+// `HttpRule` that maps the gRPC method to a REST endpoint, achieving the same
+// effect as the proto annotation. This can be particularly useful if you
+// have a proto that is reused in multiple services. Note that any transcoding
+// specified in the service config will override any matching transcoding
+// configuration in the proto.
+//
+// Example:
+//
+//     http:
+//       rules:
+//         # Selects a gRPC method and applies HttpRule to it.
+//         - selector: example.v1.Messaging.GetMessage
+//           get: /v1/messages/{message_id}/{sub.subfield}
+//
+// ## Special notes
+//
+// When gRPC Transcoding is used to map a gRPC to JSON REST endpoints, the
+// proto to JSON conversion must follow the [proto3
+// specification](https://developers.google.com/protocol-buffers/docs/proto3#json).
+//
+// While the single segment variable follows the semantics of
+// [RFC 6570](https://tools.ietf.org/html/rfc6570) Section 3.2.2 Simple String
+// Expansion, the multi segment variable **does not** follow RFC 6570 Section
+// 3.2.3 Reserved Expansion. The reason is that the Reserved Expansion
+// does not expand special characters like `?` and `#`, which would lead
+// to invalid URLs. As the result, gRPC Transcoding uses a custom encoding
+// for multi segment variables.
+//
+// The path variables **must not** refer to any repeated or mapped field,
+// because client libraries are not capable of handling such variable expansion.
+//
+// The path variables **must not** capture the leading "/" character. The reason
+// is that the most common use case "{var}" does not capture the leading "/"
+// character. For consistency, all path variables must share the same behavior.
+//
+// Repeated message fields must not be mapped to URL query parameters, because
+// no client library can support such complicated mapping.
+//
+// If an API needs to use a JSON array for request or response body, it can map
+// the request or response body to a repeated field. However, some gRPC
+// Transcoding implementations may not support this feature.
+message HttpRule {
+  // Selects a method to which this rule applies.
+  //
+  // Refer to [selector][google.api.DocumentationRule.selector] for syntax details.
+  string selector = 1;
+
+  // Determines the URL pattern is matched by this rules. This pattern can be
+  // used with any of the {get|put|post|delete|patch} methods. A custom method
+  // can be defined using the 'custom' field.
+  oneof pattern {
+    // Maps to HTTP GET. Used for listing and getting information about
+    // resources.
+    string get = 2;
+
+    // Maps to HTTP PUT. Used for replacing a resource.
+    string put = 3;
+
+    // Maps to HTTP POST. Used for creating a resource or performing an action.
+    string post = 4;
+
+    // Maps to HTTP DELETE. Used for deleting a resource.
+    string delete = 5;
+
+    // Maps to HTTP PATCH. Used for updating a resource.
+    string patch = 6;
+
+    // The custom pattern is used for specifying an HTTP method that is not
+    // included in the `pattern` field, such as HEAD, or "*" to leave the
+    // HTTP method unspecified for this rule. The wild-card rule is useful
+    // for services that provide content to Web (HTML) clients.
+    CustomHttpPattern custom = 8;
+  }
+
+  // The name of the request field whose value is mapped to the HTTP request
+  // body, or `*` for mapping all request fields not captured by the path
+  // pattern to the HTTP body, or omitted for not having any HTTP request body.
+  //
+  // NOTE: the referred field must be present at the top-level of the request
+  // message type.
+  string body = 7;
+
+  // Optional. The name of the response field whose value is mapped to the HTTP
+  // response body. When omitted, the entire response message will be used
+  // as the HTTP response body.
+  //
+  // NOTE: The referred field must be present at the top-level of the response
+  // message type.
+  string response_body = 12;
+
+  // Additional HTTP bindings for the selector. Nested bindings must
+  // not contain an `additional_bindings` field themselves (that is,
+  // the nesting may only be one level deep).
+  repeated HttpRule additional_bindings = 11;
+}
+
+// A custom pattern is used for defining custom HTTP verb.
+message CustomHttpPattern {
+  // The name of this custom HTTP verb.
+  string kind = 1;
+
+  // The path matched by this custom verb.
+  string path = 2;
+}
diff --git a/tests/fragments/google/cloud/extended_operations.proto b/tests/fragments/google/cloud/extended_operations.proto
new file mode 100644
index 0000000000..0a1ff92d1e
--- /dev/null
+++ b/tests/fragments/google/cloud/extended_operations.proto
@@ -0,0 +1,150 @@
+// Copyright 2021 Google LLC.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+// This file contains custom annotations that are used by GAPIC generators to
+// handle Long Running Operation methods (LRO) that are NOT compliant with
+// https://google.aip.dev/151. These annotations are public for technical
+// reasons only. Please DO NOT USE them in your protos.
+syntax = "proto3";
+
+package google.cloud;
+
+import "google/protobuf/descriptor.proto";
+
+option go_package = "google.golang.org/genproto/googleapis/cloud/extendedops;extendedops";
+option java_multiple_files = true;
+option java_outer_classname = "ExtendedOperationsProto";
+option java_package = "com.google.cloud";
+option objc_class_prefix = "GAPI";
+
+// FieldOptions to match corresponding fields in the initial request,
+// polling request and operation response messages.
+//
+// Example:
+//
+// In an API-specific operation message:
+//
+//     message MyOperation {
+//       string http_error_message = 1 [(operation_field) = ERROR_MESSAGE];
+//       int32 http_error_status_code = 2 [(operation_field) = ERROR_CODE];
+//       string id = 3 [(operation_field) = NAME];
+//       Status status = 4 [(operation_field) = STATUS];
+//     }
+//
+// In a polling request message (the one which is used to poll for an LRO
+// status):
+//
+//     message MyPollingRequest {
+//       string operation = 1 [(operation_response_field) = "id"];
+//       string project = 2;
+//       string region = 3;
+//     }
+//
+// In an initial request message (the one which starts an LRO):
+//
+//    message MyInitialRequest {
+//      string my_project = 2 [(operation_request_field) = "project"];
+//      string my_region = 3 [(operation_request_field) = "region"];
+//    }
+//
+extend google.protobuf.FieldOptions {
+  // A field annotation that maps fields in an API-specific Operation object to
+  // their standard counterparts in google.longrunning.Operation. See
+  // OperationResponseMapping enum definition.
+  OperationResponseMapping operation_field = 1149;
+
+  // A field annotation that maps fields in the initial request message
+  // (the one which started the LRO) to their counterparts in the polling
+  // request message. For non-standard LRO, the polling response may be missing
+  // some of the information needed to make a subsequent polling request. The
+  // missing information (for example, project or region ID) is contained in the
+  // fields of the initial request message that this annotation must be applied
+  // to. The string value of the annotation corresponds to the name of the
+  // counterpart field in the polling request message that the annotated field's
+  // value will be copied to.
+  string operation_request_field = 1150;
+
+  // A field annotation that maps fields in the polling request message to their
+  // counterparts in the initial and/or polling response message. The initial
+  // and the polling methods return an API-specific Operation object. Some of
+  // the fields from that response object must be reused in the subsequent
+  // request (like operation name/ID) to fully identify the polled operation.
+  // This annotation must be applied to the fields in the polling request
+  // message, the string value of the annotation must correspond to the name of
+  // the counterpart field in the Operation response object whose value will be
+  // copied to the annotated field.
+  string operation_response_field = 1151;
+}
+
+// MethodOptions to identify the actual service and method used for operation
+// status polling.
+//
+// Example:
+//
+// In a method, which starts an LRO:
+//
+//     service MyService {
+//       rpc Foo(MyInitialRequest) returns (MyOperation) {
+//         option (operation_service) = "MyPollingService";
+//       }
+//     }
+//
+// In a polling method:
+//
+//     service MyPollingService {
+//       rpc Get(MyPollingRequest) returns (MyOperation) {
+//         option (operation_polling_method) = true;
+//       }
+//     }
+extend google.protobuf.MethodOptions {
+  // A method annotation that maps an LRO method (the one which starts an LRO)
+  // to the service, which will be used to poll for the operation status. The
+  // annotation must be applied to the method which starts an LRO, the string
+  // value of the annotation must correspond to the name of the service used to
+  // poll for the operation status.
+  string operation_service = 1249;
+
+  // A method annotation that marks methods that can be used for polling
+  // operation status (e.g. the MyPollingService.Get(MyPollingRequest) method).
+  bool operation_polling_method = 1250;
+}
+
+// An enum to be used to mark the essential (for polling) fields in an
+// API-specific Operation object. A custom Operation object may contain many
+// different fields, but only few of them are essential to conduct a successful
+// polling process.
+enum OperationResponseMapping {
+  // Do not use.
+  UNDEFINED = 0;
+
+  // A field in an API-specific (custom) Operation object which carries the same
+  // meaning as google.longrunning.Operation.name.
+  NAME = 1;
+
+  // A field in an API-specific (custom) Operation object which carries the same
+  // meaning as google.longrunning.Operation.done. If the annotated field is of
+  // an enum type, `annotated_field_name == EnumType.DONE` semantics should be
+  // equivalent to `Operation.done == true`. If the annotated field is of type
+  // boolean, then it should follow the same semantics as Operation.done.
+  // Otherwise, a non-empty value should be treated as `Operation.done == true`.
+  STATUS = 2;
+
+  // A field in an API-specific (custom) Operation object which carries the same
+  // meaning as google.longrunning.Operation.error.code.
+  ERROR_CODE = 3;
+
+  // A field in an API-specific (custom) Operation object which carries the same
+  // meaning as google.longrunning.Operation.error.message.
+  ERROR_MESSAGE = 4;
+}
diff --git a/tests/fragments/test_diregapic_forwardcompat_lro.proto b/tests/fragments/test_diregapic_forwardcompat_lro.proto
new file mode 100644
index 0000000000..f7206a7c6b
--- /dev/null
+++ b/tests/fragments/test_diregapic_forwardcompat_lro.proto
@@ -0,0 +1,101 @@
+// Copyright (C) 2021  Google LLC
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+syntax = "proto3";
+
+package google.fragment;
+
+import "google/api/annotations.proto";
+import "google/api/client.proto";
+import "google/cloud/extended_operations.proto";
+
+message Operation {
+  enum Status {
+    // A value indicating that the enum field is not set.
+    UNDEFINED_STATUS = 0;
+    DONE = 2104194;
+    PENDING = 35394935;
+    RUNNING = 121282975;
+  }
+
+  optional string name = 1 [(google.cloud.operation_field) = NAME];
+  optional string http_error_message = 202521945 [(google.cloud.operation_field) = ERROR_MESSAGE];
+  optional int32 http_error_status_code = 312345196 [(google.cloud.operation_field) = ERROR_CODE];
+  optional Status status = 181260274 [(google.cloud.operation_field) = STATUS];
+}
+
+message GetExpensiveComputationRequest {
+  string name = 1;
+  string computation = 2;
+  string fragment = 3;
+}
+
+message DeleteExpensiveComputationRequest {
+  string computation = 1;
+  string fragment = 2;
+}
+
+message DeleteOperationResponse {
+  bool success = 1;
+}
+
+message StartExpensiveComputationRequest {
+  string computation = 1;
+  string fragment = 2;
+}
+
+message StartCheapComputationRequest {
+  string computation = 1;
+  string fragment = 2;
+}
+
+message StartCheapComputationResponse {}
+
+// Note: the name is a hint, but this is the de facto Operations client.
+// It should _not_ generate methods with disambiguated names.
+service DefactoOperationsClient {
+  option (google.api.default_host) = "my.example.com";
+
+  rpc Get(GetExpensiveComputationRequest) returns (Operation) {
+    option (google.cloud.operation_polling_method) = true;
+    option (google.api.http) = {
+      get: "/expensive/v1/computations/{computation}/fragments/{fragment}"
+    };
+  }
+
+
+  rpc Delete(DeleteExpensiveComputationRequest) returns (DeleteOperationResponse) {
+    option (google.api.http) = {
+      delete: "/expensive/v1/computations/{computation}/fragments/{fragment}"
+    };
+
+  }
+}
+
+
+service DoStuff {
+  option (google.api.default_host) = "my.example.com";
+
+  rpc StartExpensiveComputation(StartExpensiveComputationRequest) returns (Operation) {
+    option (google.cloud.operation_service) = "GlobalOperations";
+    option (google.api.http) = {
+      post: "/expensive/v1/computations/{computation}/fragments/{fragment}"
+    };
+  }
+
+  rpc StartCheapComputation(StartCheapComputationRequest) returns (StartCheapComputationResponse) {
+    option (google.api.http) = {
+      post: "/expensive/v1/computations/{computation}/fragments/{fragment}";
+    };
+  }
+}
diff --git a/tests/unit/schema/wrappers/test_message.py b/tests/unit/schema/wrappers/test_message.py
index f0b5d61166..da21f66ebd 100644
--- a/tests/unit/schema/wrappers/test_message.py
+++ b/tests/unit/schema/wrappers/test_message.py
@@ -20,6 +20,7 @@
 
 from google.api import field_behavior_pb2
 from google.api import resource_pb2
+from google.cloud import extended_operations_pb2 as ex_ops_pb2
 from google.protobuf import descriptor_pb2
 
 from gapic.schema import naming
@@ -307,3 +308,76 @@ def test_required_fields():
     )
 
     assert set(request.required_fields) == {mass_kg, length_m, color}
+
+
+def test_is_diregapic_operation():
+    T = descriptor_pb2.FieldDescriptorProto.Type
+
+    # Canonical Operation
+
+    operation = make_message(
+        name="Operation",
+        fields=(
+            make_field(name=name, type=T.Value("TYPE_STRING"), number=i)
+            for i, name in enumerate(("name", "status", "error_code", "error_message"), start=1)
+        )
+    )
+    for f in operation.field:
+        options = descriptor_pb2.FieldOptions()
+        # Note: The field numbers were carefully chosen to be the corresponding enum values.
+        options.Extensions[ex_ops_pb2.operation_field] = f.number
+        f.options.MergeFrom(options)
+
+    assert operation.is_diregapic_operation
+
+    # Missing a required field
+
+    missing = make_message(
+        name="Operation",
+        fields=(
+            make_field(name=name, type=T.Value("TYPE_STRING"), number=i)
+            # Missing error_message
+            for i, name in enumerate(("name", "status", "error_code"), start=1)
+        )
+    )
+    for f in missing.field:
+        options = descriptor_pb2.FieldOptions()
+        # Note: The field numbers were carefully chosen to be the corresponding enum values.
+        options.Extensions[ex_ops_pb2.operation_field] = f.number
+        f.options.MergeFrom(options)
+
+    assert not missing.is_diregapic_operation
+
+    # Named incorrectly
+
+    my_message = make_message(
+        name="MyMessage",
+        fields=(
+            make_field(name=name, type=T.Value("TYPE_STRING"), number=i)
+            for i, name in enumerate(("name", "status", "error_code", "error_message"), start=1)
+        )
+    )
+    for f in my_message.field:
+        options = descriptor_pb2.FieldOptions()
+        options.Extensions[ex_ops_pb2.operation_field] = f.number
+        f.options.MergeFrom(options)
+
+    assert not my_message.is_diregapic_operation
+
+    # Duplicated annotation
+    for mapping in range(1, 5):
+        duplicate = make_message(
+            name="Operation",
+            fields=(
+                make_field(name=name, type=T.Value("TYPE_STRING"), number=i)
+                for i, name in enumerate(("name", "status", "error_code", "error_message"), start=1)
+            )
+        )
+        for f in duplicate.field:
+            options = descriptor_pb2.FieldOptions()
+            # All set to the same value
+            options.Extensions[ex_ops_pb2.operation_field] = mapping
+            f.options.MergeFrom(options)
+
+        with pytest.raises(TypeError):
+            duplicate.is_diregapic_operation
diff --git a/tests/unit/schema/wrappers/test_method.py b/tests/unit/schema/wrappers/test_method.py
index 42198e09eb..774f81b172 100644
--- a/tests/unit/schema/wrappers/test_method.py
+++ b/tests/unit/schema/wrappers/test_method.py
@@ -19,6 +19,7 @@
 
 from google.api import field_behavior_pb2
 from google.api import http_pb2
+from google.cloud import extended_operations_pb2 as ex_ops_pb2
 from google.protobuf import descriptor_pb2
 
 from gapic.schema import metadata
@@ -718,3 +719,60 @@ def test_flattened_oneof_fields():
     }
     actual = method.flattened_field_to_key
     assert expected == actual
+
+
+def test_is_operation_polling_method():
+    T = descriptor_pb2.FieldDescriptorProto.Type
+
+    operation = make_message(
+        name="Operation",
+        fields=(
+            make_field(name=name, type=T.Value("TYPE_STRING"), number=i)
+            for i, name in enumerate(("name", "status", "error_code", "error_message"), start=1)
+        ),
+    )
+    for f in operation.field:
+        options = descriptor_pb2.FieldOptions()
+        # Note: The field numbers were carefully chosen to be the corresponding enum values.
+        options.Extensions[ex_ops_pb2.operation_field] = f.number
+        f.options.MergeFrom(options)
+
+    request = make_message(
+        name="GetOperation",
+        fields=[
+            make_field(name="name", type=T.Value("TYPE_STRING"), number=1)
+        ],
+    )
+
+    # Correct positive
+    options = descriptor_pb2.MethodOptions()
+    options.Extensions[ex_ops_pb2.operation_polling_method] = True
+    polling_method = make_method(
+        name="Get",
+        input_message=request,
+        output_message=operation,
+        options=options,
+    )
+
+    assert polling_method.is_operation_polling_method
+
+    # Normal method that returns operation
+    normal_method = make_method(
+        name="Get",
+        input_message=request,
+        output_message=operation,
+    )
+
+    assert not normal_method.is_operation_polling_method
+
+    # Method with invalid options combination
+    response = make_message(name="Response", fields=[make_field(name="name")])
+
+    invalid_method = make_method(
+        name="Get",
+        input_message=request,
+        output_message=response,
+        options=options,        # Reuse options from the actual polling method
+    )
+
+    assert not invalid_method.is_operation_polling_method
diff --git a/tests/unit/schema/wrappers/test_service.py b/tests/unit/schema/wrappers/test_service.py
index f93ddc814e..7cd41799ed 100644
--- a/tests/unit/schema/wrappers/test_service.py
+++ b/tests/unit/schema/wrappers/test_service.py
@@ -17,6 +17,7 @@
 import typing
 
 from google.api import resource_pb2
+from google.cloud import extended_operations_pb2 as ex_ops_pb2
 from google.protobuf import descriptor_pb2
 
 from gapic.schema import imp
@@ -524,3 +525,133 @@ def test_resource_response():
     expected = {squid_resource, clam_resource}
     actual = mollusc_service.resource_messages
     assert expected == actual
+
+
+def test_operation_polling_method():
+    T = descriptor_pb2.FieldDescriptorProto.Type
+
+    operation = make_message(
+        name="Operation",
+        fields=(
+            make_field(name=name, type=T.Value("TYPE_STRING"), number=i)
+            for i, name in enumerate(("name", "status", "error_code", "error_message"), start=1)
+        ),
+    )
+    for f in operation.field:
+        options = descriptor_pb2.FieldOptions()
+        # Note: The field numbers were carefully chosen to be the corresponding enum values.
+        options.Extensions[ex_ops_pb2.operation_field] = f.number
+        f.options.MergeFrom(options)
+
+    request = make_message(
+        name="GetOperation",
+        fields=[
+            make_field(name="name", type=T.Value("TYPE_STRING"), number=1)
+        ],
+    )
+
+    options = descriptor_pb2.MethodOptions()
+    options.Extensions[ex_ops_pb2.operation_polling_method] = True
+    polling_method = make_method(
+        name="Get",
+        input_message=request,
+        output_message=operation,
+        options=options,
+    )
+
+    # Even though polling_method returns an Operation, it isn't an LRO
+    ops_service = make_service(
+        name="CustomOperations",
+        methods=[
+            polling_method,
+            make_method(
+                name="Delete",
+                input_message=make_message(name="Input"),
+                output_message=make_message("Output"),
+            ),
+        ],
+    )
+
+    assert ops_service.custom_polling_method == polling_method
+
+    # Methods are LROs, so they are not polling methods
+    user_service = make_service(
+        name="ComputationStarter",
+        methods=[
+            make_method(
+                name="Start",
+                input_message=make_message(name="StartRequest"),
+                output_message=operation,
+            ),
+        ],
+    )
+
+    assert not user_service.custom_polling_method
+
+
+def test_diregapic_lro_detection():
+    T = descriptor_pb2.FieldDescriptorProto.Type
+
+    operation = make_message(
+        name="Operation",
+        fields=(
+            make_field(name=name, type=T.Value("TYPE_STRING"), number=i)
+            for i, name in enumerate(("name", "status", "error_code", "error_message"), start=1)
+        ),
+    )
+    for f in operation.field:
+        options = descriptor_pb2.FieldOptions()
+        # Note: The field numbers were carefully chosen to be the corresponding enum values.
+        options.Extensions[ex_ops_pb2.operation_field] = f.number
+        f.options.MergeFrom(options)
+
+    request = make_message(
+        name="GetOperation",
+        fields=[
+            make_field(name="name", type=T.Value("TYPE_STRING"), number=1)
+        ],
+    )
+
+    options = descriptor_pb2.MethodOptions()
+    options.Extensions[ex_ops_pb2.operation_polling_method] = True
+    polling_method = make_method(
+        name="Get",
+        input_message=request,
+        output_message=operation,
+        options=options,
+    )
+
+    ops_service = make_service(
+        name="CustomOperations",
+        methods=[
+            polling_method,
+            make_method(
+                name="Delete",
+                input_message=make_message(name="Input"),
+                output_message=make_message("Output"),
+            ),
+        ],
+    )
+
+    assert not polling_method.operation_service
+
+    # Methods are LROs, so they are not polling methods
+    lro_opts = descriptor_pb2.MethodOptions()
+    lro_opts.Extensions[ex_ops_pb2.operation_service] = "CustomOperations"
+    lro = make_method(
+        name="Start",
+        input_message=make_message(name="StartRequest"),
+        output_message=operation,
+        options=lro_opts,
+    )
+    user_service = make_service(
+        name="ComputationStarter",
+        methods=[
+            lro,
+        ],
+    )
+
+    # Note: we can't have the operation_serivce property point to the actual operation service
+    # because Service objects can't perform the lookup.
+    # Instead we kick that can to the API object and make it do the lookup and verification.
+    assert lro.operation_service == "CustomOperations"