-
Notifications
You must be signed in to change notification settings - Fork 1.1k
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
BIgtable: 11. Implement ReadRows retries #2986
Merged
garrettjonesgoogle
merged 9 commits into
googleapis:master
from
igorbernstein2:11-impl-readrows-retries
Mar 6, 2018
Merged
Changes from all commits
Commits
Show all changes
9 commits
Select commit
Hold shift + click to select a range
351aae9
BIgtable: 11. Implement ReadRows retries
igorbernstein2 fd8ced6
add default case
igorbernstein2 d1b8d8a
fix scope for testing dep
igorbernstein2 afe9321
codacy fixes
igorbernstein2 d8887c4
address feedback
igorbernstein2 d0aa23a
handle retries of fulfilled requests
igorbernstein2 a8c75c3
re-organize code: move row limit setting after the edge case, use loc…
igorbernstein2 92b4a71
address feedback
igorbernstein2 003d88e
fix javadoc link
igorbernstein2 File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
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
183 changes: 183 additions & 0 deletions
183
...main/java/com/google/cloud/bigtable/data/v2/stub/readrows/ReadRowsResumptionStrategy.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,183 @@ | ||
| /* | ||
| * Copyright 2018 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 | ||
| * | ||
| * https://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.bigtable.data.v2.stub.readrows; | ||
|
|
||
| import com.google.api.core.InternalApi; | ||
| import com.google.api.gax.retrying.StreamResumptionStrategy; | ||
| import com.google.bigtable.v2.ReadRowsRequest; | ||
| import com.google.bigtable.v2.ReadRowsRequest.Builder; | ||
| import com.google.bigtable.v2.RowRange; | ||
| import com.google.bigtable.v2.RowSet; | ||
| import com.google.cloud.bigtable.data.v2.internal.ByteStringComparator; | ||
| import com.google.cloud.bigtable.data.v2.models.RowAdapter; | ||
| import com.google.common.base.Preconditions; | ||
| import com.google.protobuf.ByteString; | ||
|
|
||
| /** | ||
| * An implementation of a {@link StreamResumptionStrategy} for merged rows. This class tracks the | ||
| * last complete row seen and upon retry can build a request to resume the stream from where it left | ||
| * off. | ||
| * | ||
| * <p>This class is considered an internal implementation detail and not meant to be used by | ||
| * applications. | ||
| */ | ||
| @InternalApi | ||
| public class ReadRowsResumptionStrategy<RowT> | ||
| implements StreamResumptionStrategy<ReadRowsRequest, RowT> { | ||
| private final RowAdapter<RowT> rowAdapter; | ||
| private ByteString lastKey = ByteString.EMPTY; | ||
| // Number of rows processed excluding Marker row. | ||
| private long numProcessed; | ||
|
|
||
| public ReadRowsResumptionStrategy(RowAdapter<RowT> rowAdapter) { | ||
| this.rowAdapter = rowAdapter; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean canResume() { | ||
| return true; | ||
| } | ||
|
|
||
| @Override | ||
| public StreamResumptionStrategy<ReadRowsRequest, RowT> createNew() { | ||
| return new ReadRowsResumptionStrategy<>(rowAdapter); | ||
| } | ||
|
|
||
| @Override | ||
| public void onProgress(RowT response) { | ||
| // Last key can come from both the last processed row key and a synthetic row marker. The | ||
| // synthetic row marker is emitted when the server has read a lot of data that was filtered out. | ||
| // The row marker can be used to trim the start of the scan, but does not contribute to the row | ||
| // limit. | ||
| lastKey = rowAdapter.getKey(response); | ||
| if (!rowAdapter.isScanMarkerRow(response)) { | ||
| // Only real rows count towards the rows limit. | ||
| numProcessed++; | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * {@inheritDoc} | ||
| * | ||
| * <p>Given a request, this implementation will narrow that request to exclude all row keys and | ||
| * ranges that would produce rows that come before {@link #lastKey}. Furthermore this | ||
| * implementation takes care to update the row limit of the request to account for all of the | ||
| * received rows. | ||
| */ | ||
| @Override | ||
| public ReadRowsRequest getResumeRequest(ReadRowsRequest request) { | ||
| // An empty lastKey means that we have not successfully read the first row, | ||
| // so resume with the original request object. | ||
| if (lastKey.isEmpty()) { | ||
| return request; | ||
| } | ||
|
|
||
| ReadRowsRequest originalRequest = request; | ||
|
|
||
| // Special case: empty query implies full table scan, so make this explicit by adding an | ||
| // unbounded range to the request | ||
| if (request.getRows().getRowKeysList().isEmpty() | ||
| && request.getRows().getRowRangesList().isEmpty()) { | ||
|
|
||
| originalRequest = | ||
| request | ||
| .toBuilder() | ||
| .setRows(RowSet.newBuilder().addRowRanges(RowRange.getDefaultInstance())) | ||
| .build(); | ||
| } | ||
|
|
||
| // Start building the resume request. The keys & ranges are cleared and will be recomputed. | ||
| Builder builder = originalRequest.toBuilder(); | ||
| builder.clearRows(); | ||
|
|
||
| RowSet.Builder rowSetBuilder = RowSet.newBuilder(); | ||
|
|
||
| for (ByteString key : originalRequest.getRows().getRowKeysList()) { | ||
| if (ByteStringComparator.INSTANCE.compare(key, lastKey) > 0) { | ||
| rowSetBuilder.addRowKeys(key); | ||
| } | ||
| } | ||
|
|
||
| for (RowRange rowRange : originalRequest.getRows().getRowRangesList()) { | ||
| RowRange.Builder rowRangeBuilder = RowRange.newBuilder(); | ||
|
|
||
| switch (rowRange.getEndKeyCase()) { | ||
| case END_KEY_CLOSED: | ||
| if (ByteStringComparator.INSTANCE.compare(rowRange.getEndKeyClosed(), lastKey) > 0) { | ||
| rowRangeBuilder.setEndKeyClosed(rowRange.getEndKeyClosed()); | ||
| } else { | ||
| continue; | ||
| } | ||
| break; | ||
| case END_KEY_OPEN: | ||
| if (ByteStringComparator.INSTANCE.compare(rowRange.getEndKeyOpen(), lastKey) > 0) { | ||
| rowRangeBuilder.setEndKeyOpen(rowRange.getEndKeyOpen()); | ||
| } else { | ||
| continue; | ||
| } | ||
| break; | ||
| case ENDKEY_NOT_SET: | ||
| rowRangeBuilder.clearEndKey(); | ||
| break; | ||
| default: | ||
| throw new IllegalArgumentException("Unknown endKeyCase: " + rowRange.getEndKeyCase()); | ||
| } | ||
|
|
||
| switch (rowRange.getStartKeyCase()) { | ||
| case STARTKEY_NOT_SET: | ||
| rowRangeBuilder.setStartKeyOpen(lastKey); | ||
| break; | ||
| case START_KEY_OPEN: | ||
| if (ByteStringComparator.INSTANCE.compare(rowRange.getStartKeyOpen(), lastKey) < 0) { | ||
| rowRangeBuilder.setStartKeyOpen(lastKey); | ||
| } else { | ||
| rowRangeBuilder.setStartKeyOpen(rowRange.getStartKeyOpen()); | ||
| } | ||
| break; | ||
| case START_KEY_CLOSED: | ||
| if (ByteStringComparator.INSTANCE.compare(rowRange.getStartKeyClosed(), lastKey) <= 0) { | ||
| rowRangeBuilder.setStartKeyOpen(lastKey); | ||
| } else { | ||
| rowRangeBuilder.setStartKeyClosed(rowRange.getStartKeyClosed()); | ||
| } | ||
| break; | ||
| default: | ||
| throw new IllegalArgumentException("Unknown startKeyCase: " + rowRange.getStartKeyCase()); | ||
| } | ||
| rowSetBuilder.addRowRanges(rowRangeBuilder.build()); | ||
| } | ||
|
|
||
| // Edge case: retrying a fulfilled request. | ||
| // A fulfilled request is one that has had all of its row keys and ranges fulfilled, or if it | ||
| // had a row limit, has seen enough rows. These requests are replaced with a marker request that | ||
| // will be handled by ReadRowsRetryCompletedCallable. See docs in ReadRowsRetryCompletedCallable | ||
| // for more details. | ||
| if ((rowSetBuilder.getRowRangesCount() == 0 && rowSetBuilder.getRowKeysCount() == 0) | ||
| || (originalRequest.getRowsLimit() > 0 && originalRequest.getRowsLimit() == numProcessed)) { | ||
| return ReadRowsRetryCompletedCallable.FULFILLED_REQUEST_MARKER; | ||
| } | ||
|
|
||
| if (originalRequest.getRowsLimit() > 0) { | ||
| Preconditions.checkState( | ||
| originalRequest.getRowsLimit() > numProcessed, | ||
| "Detected too many rows for the current row limit during a retry."); | ||
| builder.setRowsLimit(originalRequest.getRowsLimit() - numProcessed); | ||
| } | ||
|
|
||
| builder.setRows(rowSetBuilder.build()); | ||
| return builder.build(); | ||
| } | ||
| } |
72 changes: 72 additions & 0 deletions
72
.../java/com/google/cloud/bigtable/data/v2/stub/readrows/ReadRowsRetryCompletedCallable.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,72 @@ | ||
| /* | ||
| * Copyright 2018 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 | ||
| * | ||
| * https://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.bigtable.data.v2.stub.readrows; | ||
|
|
||
| import com.google.api.core.InternalApi; | ||
| import com.google.api.gax.rpc.ApiCallContext; | ||
| import com.google.api.gax.rpc.ResponseObserver; | ||
| import com.google.api.gax.rpc.ServerStreamingCallable; | ||
| import com.google.api.gax.rpc.StreamController; | ||
| import com.google.bigtable.v2.ReadRowsRequest; | ||
|
|
||
| /** | ||
| * This callable addresses edge case of a ReadRows stream receiving all of the rows, but receiving a | ||
| * retryable error status instead of an OK. If a retry attempt is scheduled, then it should return | ||
| * an OK response. | ||
| * | ||
| * <p>This callable works in tandem with {@link ReadRowsResumptionStrategy}, which will send a | ||
| * {@link #FULFILLED_REQUEST_MARKER} to be processed by this callable. Upon receiving the {@link | ||
| * #FULFILLED_REQUEST_MARKER}, this callable will promptly notify the {@link ResponseObserver} that | ||
| * the stream has been successfully compeleted. | ||
| * | ||
| * <p>This class is considered an internal implementation detail and not meant to be used by | ||
| * applications. | ||
| */ | ||
| @InternalApi | ||
| public final class ReadRowsRetryCompletedCallable<RowT> | ||
| extends ServerStreamingCallable<ReadRowsRequest, RowT> { | ||
| static final ReadRowsRequest FULFILLED_REQUEST_MARKER = | ||
| ReadRowsRequest.newBuilder().setRowsLimit(-1).build(); | ||
|
|
||
| private final ServerStreamingCallable<ReadRowsRequest, RowT> inner; | ||
|
|
||
| public ReadRowsRetryCompletedCallable(ServerStreamingCallable<ReadRowsRequest, RowT> inner) { | ||
| this.inner = inner; | ||
| } | ||
|
|
||
| @Override | ||
| public void call( | ||
| ReadRowsRequest request, ResponseObserver<RowT> responseObserver, ApiCallContext context) { | ||
|
|
||
| if (request == FULFILLED_REQUEST_MARKER) { | ||
| responseObserver.onStart(new DummyController()); | ||
| responseObserver.onComplete(); | ||
| } else { | ||
| inner.call(request, responseObserver, context); | ||
| } | ||
| } | ||
|
|
||
| private static class DummyController implements StreamController { | ||
| @Override | ||
| public void cancel() {} | ||
|
|
||
| @Override | ||
| public void disableAutoInboundFlowControl() {} | ||
|
|
||
| @Override | ||
| public void request(int count) {} | ||
| } | ||
| } |
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
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
This comment was marked as spam.
Sorry, something went wrong.
This comment was marked as spam.
Sorry, something went wrong.