Skip to content

Commit

Permalink
feat: Implement internal CursorClient which will be used by kafka shi…
Browse files Browse the repository at this point in the history
…m. (#252)
  • Loading branch information
dpcollins-google committed Sep 24, 2020
1 parent 4111f09 commit eabe900
Show file tree
Hide file tree
Showing 5 changed files with 583 additions and 0 deletions.
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
/*
* 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.SubscriptionPath;
import io.grpc.StatusException;
import java.util.Map;

public interface CursorClient extends BackgroundResource {

static CursorClient create(CursorClientSettings settings) throws StatusException {
return settings.instantiate();
}

/** The Google Cloud region this client operates on. */
CloudRegion region();

/**
* List the cursors for a given subscription.
*
* @param path The subscription to list cursors for.
* @return A future holding the map of Partition to Offset of the cursors.
*/
ApiFuture<Map<Partition, Offset>> listPartitionCursors(SubscriptionPath path);

/**
* Commit a single cursor.
*
* @param path The subscription to commit a cursor for.
* @param partition The partition to commit a cursor for.
* @param offset The offset to commit.
* @return A future for the operation's completion.
*/
ApiFuture<Void> commitCursor(SubscriptionPath path, Partition partition, Offset offset);

/**
* Tear down this admin client.
*
* @throws StatusException on a failure to properly terminate.
*/
@Override
void close() throws StatusException;
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,144 @@
/*
* 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.api.gax.core.ExecutorAsBackgroundResource;
import com.google.api.gax.retrying.RetrySettings;
import com.google.api.gax.retrying.RetryingExecutor;
import com.google.cloud.pubsublite.CloudRegion;
import com.google.cloud.pubsublite.Offset;
import com.google.cloud.pubsublite.Partition;
import com.google.cloud.pubsublite.SubscriptionPath;
import com.google.cloud.pubsublite.proto.CommitCursorRequest;
import com.google.cloud.pubsublite.proto.CommitCursorResponse;
import com.google.cloud.pubsublite.proto.Cursor;
import com.google.cloud.pubsublite.proto.CursorServiceGrpc.CursorServiceBlockingStub;
import com.google.cloud.pubsublite.proto.ListPartitionCursorsRequest;
import com.google.cloud.pubsublite.proto.ListPartitionCursorsResponse;
import com.google.cloud.pubsublite.proto.PartitionCursor;
import com.google.common.collect.ImmutableMap;
import io.grpc.StatusException;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class CursorClientImpl implements BackgroundResource, CursorClient {
private final ExecutorAsBackgroundResource executorResource;
private final CloudRegion region;
private final CursorServiceBlockingStub stub;
private final RetryingExecutor<Map<Partition, Offset>> listRetryingExecutor;
private final RetryingExecutor<Void> voidRetryingExecutor;

public CursorClientImpl(
CloudRegion region, CursorServiceBlockingStub stub, RetrySettings retrySettings) {
this(
region,
stub,
retrySettings,
// TODO: Consider allowing tuning in the future.
Executors.newScheduledThreadPool(6));
}

private CursorClientImpl(
CloudRegion region,
CursorServiceBlockingStub stub,
RetrySettings retrySettings,
ScheduledExecutorService executor) {
this.executorResource = new ExecutorAsBackgroundResource(executor);
this.region = region;
this.stub = stub;
this.listRetryingExecutor = RetryingExecutorUtil.retryingExecutor(retrySettings, executor);
this.voidRetryingExecutor = RetryingExecutorUtil.retryingExecutor(retrySettings, executor);
}

@Override
public CloudRegion region() {
return region;
}

// BackgroundResource implementation.
@Override
public void shutdown() {
executorResource.shutdown();
}

@Override
public boolean isShutdown() {
return executorResource.isShutdown();
}

@Override
public boolean isTerminated() {
return executorResource.isTerminated();
}

@Override
public void shutdownNow() {
executorResource.shutdownNow();
}

@Override
public boolean awaitTermination(long duration, TimeUnit unit) throws InterruptedException {
return executorResource.awaitTermination(duration, unit);
}

@Override
public void close() throws StatusException {
try {
executorResource.close();
} catch (Exception e) {
throw ExtractStatus.toCanonical(e);
}
}

// CursorClient Implementation
@Override
public ApiFuture<Map<Partition, Offset>> listPartitionCursors(SubscriptionPath path) {
return RetryingExecutorUtil.runWithRetries(
() -> {
ListPartitionCursorsResponse response =
stub.listPartitionCursors(
ListPartitionCursorsRequest.newBuilder().setParent(path.toString()).build());
ImmutableMap.Builder<Partition, Offset> resultBuilder = ImmutableMap.builder();
for (PartitionCursor partitionCursor : response.getPartitionCursorsList()) {
resultBuilder.put(
Partition.of(partitionCursor.getPartition()),
Offset.of(partitionCursor.getCursor().getOffset()));
}
return resultBuilder.build();
},
listRetryingExecutor);
}

@Override
public ApiFuture<Void> commitCursor(SubscriptionPath path, Partition partition, Offset offset) {
return RetryingExecutorUtil.runWithRetries(
() -> {
CommitCursorResponse unusedResponse =
stub.commitCursor(
CommitCursorRequest.newBuilder()
.setSubscription(path.toString())
.setPartition(partition.value())
.setCursor(Cursor.newBuilder().setOffset(offset.value()))
.build());
return null;
},
voidRetryingExecutor);
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,67 @@
/*
* 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.gax.retrying.RetrySettings;
import com.google.auto.value.AutoValue;
import com.google.cloud.pubsublite.CloudRegion;
import com.google.cloud.pubsublite.Constants;
import com.google.cloud.pubsublite.Stubs;
import com.google.cloud.pubsublite.proto.CursorServiceGrpc;
import com.google.cloud.pubsublite.proto.CursorServiceGrpc.CursorServiceBlockingStub;
import io.grpc.StatusException;
import java.util.Optional;

@AutoValue
public abstract class CursorClientSettings {

// Required parameters.
abstract CloudRegion region();

// Optional parameters.
abstract RetrySettings retrySettings();

abstract Optional<CursorServiceBlockingStub> stub();

public static Builder newBuilder() {
return new AutoValue_CursorClientSettings.Builder()
.setRetrySettings(Constants.DEFAULT_RETRY_SETTINGS);
}

@AutoValue.Builder
public abstract static class Builder {

// Required parameters.
public abstract Builder setRegion(CloudRegion region);

public abstract Builder setRetrySettings(RetrySettings retrySettings);

// Optional parameters.
public abstract Builder setStub(CursorServiceBlockingStub stub);

public abstract CursorClientSettings build();
}

CursorClient instantiate() throws StatusException {
CursorServiceBlockingStub stub;
if (stub().isPresent()) {
stub = stub().get();
} else {
stub = Stubs.defaultStub(region(), CursorServiceGrpc::newBlockingStub);
}
return new CursorClientImpl(region(), stub, retrySettings());
}
}
Loading

0 comments on commit eabe900

Please sign in to comment.