Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: Add the TopicStats client (#179)
* feat: Add the TopicStats client Adds an interface and implementation for the TopicStatsClient. I've kept them both in internal for now since we don't yet expose an API for interacting with the CursorService. * Changes to address review comments * move comment
- Loading branch information
1 parent
f3c4f79
commit 7eb7861
Showing
10 changed files
with
642 additions
and
108 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
77 changes: 77 additions & 0 deletions
77
...d-pubsublite/src/main/java/com/google/cloud/pubsublite/internal/RetryingExecutorUtil.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,77 @@ | ||
| /* | ||
| * Copyright 2020 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. | ||
| */ | ||
| package com.google.cloud.pubsublite.internal; | ||
|
|
||
| import com.google.api.core.ApiFuture; | ||
| import com.google.api.core.NanoClock; | ||
| import com.google.api.gax.retrying.ExponentialRetryAlgorithm; | ||
| import com.google.api.gax.retrying.ResultRetryAlgorithm; | ||
| import com.google.api.gax.retrying.RetryAlgorithm; | ||
| import com.google.api.gax.retrying.RetrySettings; | ||
| import com.google.api.gax.retrying.RetryingExecutor; | ||
| import com.google.api.gax.retrying.RetryingFuture; | ||
| import com.google.api.gax.retrying.ScheduledRetryingExecutor; | ||
| import com.google.api.gax.retrying.TimedAttemptSettings; | ||
| import com.google.cloud.pubsublite.ErrorCodes; | ||
| import io.grpc.Status; | ||
| import java.util.Optional; | ||
| import java.util.concurrent.Callable; | ||
| import java.util.concurrent.ScheduledExecutorService; | ||
|
|
||
| public final class RetryingExecutorUtil { | ||
|
|
||
| private RetryingExecutorUtil() {} | ||
|
|
||
| public static <T> RetryingExecutor<T> retryingExecutor( | ||
| RetrySettings settings, ScheduledExecutorService executor) { | ||
| return new ScheduledRetryingExecutor<>(retryAlgorithm(settings), executor); | ||
| } | ||
|
|
||
| public static <T> ApiFuture<T> runWithRetries( | ||
| Callable<T> callable, RetryingExecutor<T> executor) { | ||
| RetryingFuture<T> retryingFuture = executor.createFuture(callable); | ||
| retryingFuture.setAttemptFuture(executor.submit(retryingFuture)); | ||
| return retryingFuture; | ||
| } | ||
|
|
||
| private static <T> RetryAlgorithm<T> retryAlgorithm(RetrySettings retrySettings) { | ||
| return new RetryAlgorithm<>( | ||
| resultRetryAlgorithm(), | ||
| new ExponentialRetryAlgorithm(retrySettings, NanoClock.getDefaultClock())); | ||
| } | ||
|
|
||
| private static <T> ResultRetryAlgorithm<T> resultRetryAlgorithm() { | ||
| return new ResultRetryAlgorithm<T>() { | ||
| @Override | ||
| public TimedAttemptSettings createNextAttempt( | ||
| Throwable prevThrowable, T prevResponse, TimedAttemptSettings prevSettings) { | ||
| return null; // Null means no specific settings. | ||
| } | ||
|
|
||
| @Override | ||
| public boolean shouldRetry(Throwable prevThrowable, T prevResponse) { | ||
| if (null != prevResponse) { | ||
| return false; | ||
| } | ||
| Optional<Status> statusOr = ExtractStatus.extract(prevThrowable); | ||
| if (!statusOr.isPresent()) { | ||
| return false; // Received a non-grpc error. | ||
| } | ||
| return ErrorCodes.IsRetryable(statusOr.get().getCode()); | ||
| } | ||
| }; | ||
| } | ||
| } |
55 changes: 55 additions & 0 deletions
55
...cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/internal/TopicStatsClient.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| /* | ||
| * Copyright 2020 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. | ||
| */ | ||
| package com.google.cloud.pubsublite.internal; | ||
|
|
||
| import com.google.api.core.ApiFuture; | ||
| import com.google.api.gax.core.BackgroundResource; | ||
| import com.google.cloud.pubsublite.CloudRegion; | ||
| import com.google.cloud.pubsublite.Offset; | ||
| import com.google.cloud.pubsublite.Partition; | ||
| import com.google.cloud.pubsublite.TopicPath; | ||
| import com.google.cloud.pubsublite.proto.ComputeMessageStatsResponse; | ||
| import io.grpc.StatusException; | ||
|
|
||
| public interface TopicStatsClient extends BackgroundResource { | ||
|
|
||
| static TopicStatsClient create(TopicStatsClientSettings settings) throws StatusException { | ||
| return settings.instantiate(); | ||
| } | ||
|
|
||
| /** The Google Cloud region this client operates on. */ | ||
| CloudRegion region(); | ||
|
|
||
| /** | ||
| * Compute statistics about the messages between two cursors in a topic partition. | ||
| * | ||
| * @param path The topic to compute statistics on | ||
| * @param partition The partition to compute statistics for | ||
| * @param start The start cursor | ||
| * @param end The end cursor | ||
| * @return A future that will have either an error {@link io.grpc.StatusException} or the | ||
| * ComputeMessageStatistics on success. | ||
| */ | ||
| ApiFuture<ComputeMessageStatsResponse> computeMessageStats( | ||
| TopicPath path, Partition partition, Offset start, Offset end); | ||
|
|
||
| /** | ||
| * Tear down this admin client. | ||
| * | ||
| * @throws StatusException on a failure to properly terminate. | ||
| */ | ||
| void close() throws StatusException; | ||
| } |
Oops, something went wrong.