Skip to content

Commit 2e29ba1

Browse files
gcf-owl-bot[bot]gcf-owl-bot[bot]
authored
feat: Add SeekSubscription and Operations to API (#169)
PiperOrigin-RevId: 380660182 Source-Link: googleapis/googleapis@b601f02 Source-Link: googleapis/googleapis-gen@de07f61
1 parent f95b4cc commit 2e29ba1

10 files changed

Lines changed: 594 additions & 0 deletions

File tree

‎google/cloud/pubsublite_v1/__init__.py‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,6 +47,9 @@
4747
from .types.admin import ListTopicsResponse
4848
from .types.admin import ListTopicSubscriptionsRequest
4949
from .types.admin import ListTopicSubscriptionsResponse
50+
from .types.admin import OperationMetadata
51+
from .types.admin import SeekSubscriptionRequest
52+
from .types.admin import SeekSubscriptionResponse
5053
from .types.admin import TopicPartitions
5154
from .types.admin import UpdateReservationRequest
5255
from .types.admin import UpdateSubscriptionRequest
@@ -147,6 +150,7 @@
147150
"MessagePublishRequest",
148151
"MessagePublishResponse",
149152
"MessageResponse",
153+
"OperationMetadata",
150154
"PartitionAssignment",
151155
"PartitionAssignmentAck",
152156
"PartitionAssignmentRequest",
@@ -159,6 +163,8 @@
159163
"Reservation",
160164
"SeekRequest",
161165
"SeekResponse",
166+
"SeekSubscriptionRequest",
167+
"SeekSubscriptionResponse",
162168
"SequencedCommitCursorRequest",
163169
"SequencedCommitCursorResponse",
164170
"SequencedMessage",

‎google/cloud/pubsublite_v1/gapic_metadata.json‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -85,6 +85,11 @@
8585
"list_topics"
8686
]
8787
},
88+
"SeekSubscription": {
89+
"methods": [
90+
"seek_subscription"
91+
]
92+
},
8893
"UpdateReservation": {
8994
"methods": [
9095
"update_reservation"
@@ -180,6 +185,11 @@
180185
"list_topics"
181186
]
182187
},
188+
"SeekSubscription": {
189+
"methods": [
190+
"seek_subscription"
191+
]
192+
},
183193
"UpdateReservation": {
184194
"methods": [
185195
"update_reservation"

‎google/cloud/pubsublite_v1/services/admin_service/async_client.py‎

Lines changed: 87 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,8 @@
2626
from google.auth import credentials as ga_credentials # type: ignore
2727
from google.oauth2 import service_account # type: ignore
2828

29+
from google.api_core import operation # type: ignore
30+
from google.api_core import operation_async # type: ignore
2931
from google.cloud.pubsublite_v1.services.admin_service import pagers
3032
from google.cloud.pubsublite_v1.types import admin
3133
from google.cloud.pubsublite_v1.types import common
@@ -1084,6 +1086,91 @@ async def delete_subscription(
10841086
request, retry=retry, timeout=timeout, metadata=metadata,
10851087
)
10861088

1089+
async def seek_subscription(
1090+
self,
1091+
request: admin.SeekSubscriptionRequest = None,
1092+
*,
1093+
retry: retries.Retry = gapic_v1.method.DEFAULT,
1094+
timeout: float = None,
1095+
metadata: Sequence[Tuple[str, str]] = (),
1096+
) -> operation_async.AsyncOperation:
1097+
r"""Performs an out-of-band seek for a subscription to a
1098+
specified target, which may be timestamps or named
1099+
positions within the message backlog. Seek translates
1100+
these targets to cursors for each partition and
1101+
orchestrates subscribers to start consuming messages
1102+
from these seek cursors.
1103+
1104+
If an operation is returned, the seek has been
1105+
registered and subscribers will eventually receive
1106+
messages from the seek cursors (i.e. eventual
1107+
consistency), as long as they are using a minimum
1108+
supported client library version and not a system that
1109+
tracks cursors independently of Pub/Sub Lite (e.g.
1110+
Apache Beam, Dataflow, Spark). The seek operation will
1111+
fail for unsupported clients.
1112+
1113+
If clients would like to know when subscribers react to
1114+
the seek (or not), they can poll the operation. The seek
1115+
operation will succeed and complete once subscribers are
1116+
ready to receive messages from the seek cursors for all
1117+
partitions of the topic. This means that the seek
1118+
operation will not complete until all subscribers come
1119+
online.
1120+
1121+
If the previous seek operation has not yet completed, it
1122+
will be aborted and the new invocation of seek will
1123+
supersede it.
1124+
1125+
Args:
1126+
request (:class:`google.cloud.pubsublite_v1.types.SeekSubscriptionRequest`):
1127+
The request object. Request for SeekSubscription.
1128+
retry (google.api_core.retry.Retry): Designation of what errors, if any,
1129+
should be retried.
1130+
timeout (float): The timeout for this request.
1131+
metadata (Sequence[Tuple[str, str]]): Strings which should be
1132+
sent along with the request as metadata.
1133+
1134+
Returns:
1135+
google.api_core.operation_async.AsyncOperation:
1136+
An object representing a long-running operation.
1137+
1138+
The result type for the operation will be
1139+
:class:`google.cloud.pubsublite_v1.types.SeekSubscriptionResponse`
1140+
Response for SeekSubscription long running operation.
1141+
1142+
"""
1143+
# Create or coerce a protobuf request object.
1144+
request = admin.SeekSubscriptionRequest(request)
1145+
1146+
# Wrap the RPC method; this adds retry and timeout information,
1147+
# and friendly error handling.
1148+
rpc = gapic_v1.method_async.wrap_method(
1149+
self._client._transport.seek_subscription,
1150+
default_timeout=None,
1151+
client_info=DEFAULT_CLIENT_INFO,
1152+
)
1153+
1154+
# Certain fields should be provided within the metadata header;
1155+
# add these here.
1156+
metadata = tuple(metadata) + (
1157+
gapic_v1.routing_header.to_grpc_metadata((("name", request.name),)),
1158+
)
1159+
1160+
# Send the request.
1161+
response = await rpc(request, retry=retry, timeout=timeout, metadata=metadata,)
1162+
1163+
# Wrap the response in an operation future.
1164+
response = operation_async.from_gapic(
1165+
response,
1166+
self._client._transport.operations_client,
1167+
admin.SeekSubscriptionResponse,
1168+
metadata_type=admin.OperationMetadata,
1169+
)
1170+
1171+
# Done; return the response.
1172+
return response
1173+
10871174
async def create_reservation(
10881175
self,
10891176
request: admin.CreateReservationRequest = None,

‎google/cloud/pubsublite_v1/services/admin_service/client.py‎

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,8 @@
3030
from google.auth.exceptions import MutualTLSChannelError # type: ignore
3131
from google.oauth2 import service_account # type: ignore
3232

33+
from google.api_core import operation # type: ignore
34+
from google.api_core import operation_async # type: ignore
3335
from google.cloud.pubsublite_v1.services.admin_service import pagers
3436
from google.cloud.pubsublite_v1.types import admin
3537
from google.cloud.pubsublite_v1.types import common
@@ -1299,6 +1301,92 @@ def delete_subscription(
12991301
request, retry=retry, timeout=timeout, metadata=metadata,
13001302
)
13011303

1304+
def seek_subscription(
1305+
self,
1306+
request: admin.SeekSubscriptionRequest = None,
1307+
*,
1308+
retry: retries.Retry = gapic_v1.method.DEFAULT,
1309+
timeout: float = None,
1310+
metadata: Sequence[Tuple[str, str]] = (),
1311+
) -> operation.Operation:
1312+
r"""Performs an out-of-band seek for a subscription to a
1313+
specified target, which may be timestamps or named
1314+
positions within the message backlog. Seek translates
1315+
these targets to cursors for each partition and
1316+
orchestrates subscribers to start consuming messages
1317+
from these seek cursors.
1318+
1319+
If an operation is returned, the seek has been
1320+
registered and subscribers will eventually receive
1321+
messages from the seek cursors (i.e. eventual
1322+
consistency), as long as they are using a minimum
1323+
supported client library version and not a system that
1324+
tracks cursors independently of Pub/Sub Lite (e.g.
1325+
Apache Beam, Dataflow, Spark). The seek operation will
1326+
fail for unsupported clients.
1327+
1328+
If clients would like to know when subscribers react to
1329+
the seek (or not), they can poll the operation. The seek
1330+
operation will succeed and complete once subscribers are
1331+
ready to receive messages from the seek cursors for all
1332+
partitions of the topic. This means that the seek
1333+
operation will not complete until all subscribers come
1334+
online.
1335+
1336+
If the previous seek operation has not yet completed, it
1337+
will be aborted and the new invocation of seek will
1338+
supersede it.
1339+
1340+
Args:
1341+
request (google.cloud.pubsublite_v1.types.SeekSubscriptionRequest):
1342+
The request object. Request for SeekSubscription.
1343+
retry (google.api_core.retry.Retry): Designation of what errors, if any,
1344+
should be retried.
1345+
timeout (float): The timeout for this request.
1346+
metadata (Sequence[Tuple[str, str]]): Strings which should be
1347+
sent along with the request as metadata.
1348+
1349+
Returns:
1350+
google.api_core.operation.Operation:
1351+
An object representing a long-running operation.
1352+
1353+
The result type for the operation will be
1354+
:class:`google.cloud.pubsublite_v1.types.SeekSubscriptionResponse`
1355+
Response for SeekSubscription long running operation.
1356+
1357+
"""
1358+
# Create or coerce a protobuf request object.
1359+
# Minor optimization to avoid making a copy if the user passes
1360+
# in a admin.SeekSubscriptionRequest.
1361+
# There's no risk of modifying the input as we've already verified
1362+
# there are no flattened fields.
1363+
if not isinstance(request, admin.SeekSubscriptionRequest):
1364+
request = admin.SeekSubscriptionRequest(request)
1365+
1366+
# Wrap the RPC method; this adds retry and timeout information,
1367+
# and friendly error handling.
1368+
rpc = self._transport._wrapped_methods[self._transport.seek_subscription]
1369+
1370+
# Certain fields should be provided within the metadata header;
1371+
# add these here.
1372+
metadata = tuple(metadata) + (
1373+
gapic_v1.routing_header.to_grpc_metadata((("name", request.name),)),
1374+
)
1375+
1376+
# Send the request.
1377+
response = rpc(request, retry=retry, timeout=timeout, metadata=metadata,)
1378+
1379+
# Wrap the response in an operation future.
1380+
response = operation.from_gapic(
1381+
response,
1382+
self._transport.operations_client,
1383+
admin.SeekSubscriptionResponse,
1384+
metadata_type=admin.OperationMetadata,
1385+
)
1386+
1387+
# Done; return the response.
1388+
return response
1389+
13021390
def create_reservation(
13031391
self,
13041392
request: admin.CreateReservationRequest = None,

‎google/cloud/pubsublite_v1/services/admin_service/transports/base.py‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,11 +23,13 @@
2323
from google.api_core import exceptions as core_exceptions # type: ignore
2424
from google.api_core import gapic_v1 # type: ignore
2525
from google.api_core import retry as retries # type: ignore
26+
from google.api_core import operations_v1 # type: ignore
2627
from google.auth import credentials as ga_credentials # type: ignore
2728
from google.oauth2 import service_account # type: ignore
2829

2930
from google.cloud.pubsublite_v1.types import admin
3031
from google.cloud.pubsublite_v1.types import common
32+
from google.longrunning import operations_pb2 # type: ignore
3133
from google.protobuf import empty_pb2 # type: ignore
3234

3335
try:
@@ -196,6 +198,9 @@ def _prep_wrapped_messages(self, client_info):
196198
self.delete_subscription: gapic_v1.method.wrap_method(
197199
self.delete_subscription, default_timeout=None, client_info=client_info,
198200
),
201+
self.seek_subscription: gapic_v1.method.wrap_method(
202+
self.seek_subscription, default_timeout=None, client_info=client_info,
203+
),
199204
self.create_reservation: gapic_v1.method.wrap_method(
200205
self.create_reservation, default_timeout=None, client_info=client_info,
201206
),
@@ -218,6 +223,11 @@ def _prep_wrapped_messages(self, client_info):
218223
),
219224
}
220225

226+
@property
227+
def operations_client(self) -> operations_v1.OperationsClient:
228+
"""Return the client designed to process long-running operations."""
229+
raise NotImplementedError()
230+
221231
@property
222232
def create_topic(
223233
self,
@@ -327,6 +337,15 @@ def delete_subscription(
327337
]:
328338
raise NotImplementedError()
329339

340+
@property
341+
def seek_subscription(
342+
self,
343+
) -> Callable[
344+
[admin.SeekSubscriptionRequest],
345+
Union[operations_pb2.Operation, Awaitable[operations_pb2.Operation]],
346+
]:
347+
raise NotImplementedError()
348+
330349
@property
331350
def create_reservation(
332351
self,

0 commit comments

Comments
 (0)