@Test public void testMultiChunkCell() { CellChunk cellChunk1 = createCell("row_key1", "family", "qualifier", "v", 1, false) .toBuilder() .setValueSize("value".length()) .build(); CellChunk cellChunk2 = createCell(null, null, null, "alue", 1, true); RowMerger underTest = new RowMerger(observer); underTest.onNext(ReadRowsResponse.newBuilder().addChunks(cellChunk1).build()); underTest.onNext(ReadRowsResponse.newBuilder().addChunks(cellChunk2).build()); ByteString value = ByteString.copyFrom("value".getBytes()); FlatRow row = FlatRow.newBuilder() .withRowKey(cellChunk1.getRowKey()) .addCell( cellChunk1.getFamilyName().getValue(), cellChunk1.getQualifier().getValue(), cellChunk1.getTimestampMicros(), value) .build(); verify(observer, times(1)).onNext(eq(row)); }
@Override public final void onNext(ReadRowsResponse readRowsResponse) { if (complete) { onError(new IllegalStateException("Adding partialRow after completion")); return; } int rowsProcessed = 0; for (int i = 0; i < readRowsResponse.getChunksCount(); i++) { try { CellChunk chunk = readRowsResponse.getChunks(i); state.validateChunk(rowInProgress, lastCompletedRowKey, chunk); if (chunk.getResetRow()) { rowInProgress = null; state = RowMergerState.NewRow; continue; } if (state == RowMergerState.NewRow) { rowInProgress = new RowInProgress(); rowInProgress.updateCurrentKey(chunk); } else if (state == RowMergerState.RowInProgress) { rowInProgress.updateCurrentKey(chunk); } if (chunk.getValueSize() > 0) { rowInProgress.addPartialCellChunk(chunk); state = RowMergerState.CellInProgress; } else if (rowInProgress.hasChunkInProgess()) { rowInProgress.addPartialCellChunk(chunk); rowInProgress.completeMultiChunkCell(); state = RowMergerState.RowInProgress; } else { rowInProgress.addFullChunk(chunk); state = RowMergerState.RowInProgress; } if (chunk.getCommitRow()) { observer.onNext(rowInProgress.buildRow()); lastCompletedRowKey = rowInProgress.getRowKey(); state = RowMergerState.NewRow; rowInProgress = null; rowsProcessed++; } } catch (Throwable e) { onError(e); return; } } this.rowCountInLastMessage = rowsProcessed; }
RowMerger implements StreamObserver<ReadRowsResponse> { @Override public final void onNext(ReadRowsResponse readRowsResponse) { if (complete) { onError(new IllegalStateException("Adding partialRow after completion")); return; } int rowsProcessed = 0; for (int i = 0; i < readRowsResponse.getChunksCount(); i++) { try { CellChunk chunk = readRowsResponse.getChunks(i); state.validateChunk(rowInProgress, lastCompletedRowKey, chunk); if (chunk.getResetRow()) { rowInProgress = null; state = RowMergerState.NewRow; continue; } if (state == RowMergerState.NewRow) { rowInProgress = new RowInProgress(); rowInProgress.updateCurrentKey(chunk); } else if (state == RowMergerState.RowInProgress) { rowInProgress.updateCurrentKey(chunk); } if (chunk.getValueSize() > 0) { rowInProgress.addPartialCellChunk(chunk); state = RowMergerState.CellInProgress; } else if (rowInProgress.hasChunkInProgess()) { rowInProgress.addPartialCellChunk(chunk); rowInProgress.completeMultiChunkCell(); state = RowMergerState.RowInProgress; } else { rowInProgress.addFullChunk(chunk); state = RowMergerState.RowInProgress; } if (chunk.getCommitRow()) { observer.onNext(rowInProgress.buildRow()); lastCompletedRowKey = rowInProgress.getRowKey(); state = RowMergerState.NewRow; rowInProgress = null; rowsProcessed++; } } catch (Throwable e) { onError(e); return; } } this.rowCountInLastMessage = rowsProcessed; } }
RowMerger implements StreamObserver<ReadRowsResponse> { @Override public final void onNext(ReadRowsResponse readRowsResponse) { if (complete) { onError(new IllegalStateException("Adding partialRow after completion")); return; } int rowsProcessed = 0; for (int i = 0; i < readRowsResponse.getChunksCount(); i++) { try { CellChunk chunk = readRowsResponse.getChunks(i); state.validateChunk(rowInProgress, lastCompletedRowKey, chunk); if (chunk.getResetRow()) { rowInProgress = null; state = RowMergerState.NewRow; continue; } if (state == RowMergerState.NewRow) { rowInProgress = new RowInProgress(); rowInProgress.updateCurrentKey(chunk); } else if (state == RowMergerState.RowInProgress) { rowInProgress.updateCurrentKey(chunk); } if (chunk.getValueSize() > 0) { rowInProgress.addPartialCellChunk(chunk); state = RowMergerState.CellInProgress; } else if (rowInProgress.hasChunkInProgess()) { rowInProgress.addPartialCellChunk(chunk); rowInProgress.completeMultiChunkCell(); state = RowMergerState.RowInProgress; } else { rowInProgress.addFullChunk(chunk); state = RowMergerState.RowInProgress; } if (chunk.getCommitRow()) { observer.onNext(rowInProgress.buildRow()); lastCompletedRowKey = rowInProgress.getRowKey(); state = RowMergerState.NewRow; rowInProgress = null; rowsProcessed++; } } catch (Throwable e) { onError(e); return; } } this.rowCountInLastMessage = rowsProcessed; } RowMerger(StreamObserver<FlatRow> observer); }
RowMerger implements StreamObserver<ReadRowsResponse> { @Override public final void onNext(ReadRowsResponse readRowsResponse) { if (complete) { onError(new IllegalStateException("Adding partialRow after completion")); return; } int rowsProcessed = 0; for (int i = 0; i < readRowsResponse.getChunksCount(); i++) { try { CellChunk chunk = readRowsResponse.getChunks(i); state.validateChunk(rowInProgress, lastCompletedRowKey, chunk); if (chunk.getResetRow()) { rowInProgress = null; state = RowMergerState.NewRow; continue; } if (state == RowMergerState.NewRow) { rowInProgress = new RowInProgress(); rowInProgress.updateCurrentKey(chunk); } else if (state == RowMergerState.RowInProgress) { rowInProgress.updateCurrentKey(chunk); } if (chunk.getValueSize() > 0) { rowInProgress.addPartialCellChunk(chunk); state = RowMergerState.CellInProgress; } else if (rowInProgress.hasChunkInProgess()) { rowInProgress.addPartialCellChunk(chunk); rowInProgress.completeMultiChunkCell(); state = RowMergerState.RowInProgress; } else { rowInProgress.addFullChunk(chunk); state = RowMergerState.RowInProgress; } if (chunk.getCommitRow()) { observer.onNext(rowInProgress.buildRow()); lastCompletedRowKey = rowInProgress.getRowKey(); state = RowMergerState.NewRow; rowInProgress = null; rowsProcessed++; } } catch (Throwable e) { onError(e); return; } } this.rowCountInLastMessage = rowsProcessed; } RowMerger(StreamObserver<FlatRow> observer); static List<FlatRow> toRows(Iterable<ReadRowsResponse> responses); void clearRowInProgress(); @Override final void onNext(ReadRowsResponse readRowsResponse); Integer getRowCountInLastMessage(); ByteString getLastCompletedRowKey(); @Override void onCompleted(); @Override void onError(Throwable e); }
RowMerger implements StreamObserver<ReadRowsResponse> { @Override public final void onNext(ReadRowsResponse readRowsResponse) { if (complete) { onError(new IllegalStateException("Adding partialRow after completion")); return; } int rowsProcessed = 0; for (int i = 0; i < readRowsResponse.getChunksCount(); i++) { try { CellChunk chunk = readRowsResponse.getChunks(i); state.validateChunk(rowInProgress, lastCompletedRowKey, chunk); if (chunk.getResetRow()) { rowInProgress = null; state = RowMergerState.NewRow; continue; } if (state == RowMergerState.NewRow) { rowInProgress = new RowInProgress(); rowInProgress.updateCurrentKey(chunk); } else if (state == RowMergerState.RowInProgress) { rowInProgress.updateCurrentKey(chunk); } if (chunk.getValueSize() > 0) { rowInProgress.addPartialCellChunk(chunk); state = RowMergerState.CellInProgress; } else if (rowInProgress.hasChunkInProgess()) { rowInProgress.addPartialCellChunk(chunk); rowInProgress.completeMultiChunkCell(); state = RowMergerState.RowInProgress; } else { rowInProgress.addFullChunk(chunk); state = RowMergerState.RowInProgress; } if (chunk.getCommitRow()) { observer.onNext(rowInProgress.buildRow()); lastCompletedRowKey = rowInProgress.getRowKey(); state = RowMergerState.NewRow; rowInProgress = null; rowsProcessed++; } } catch (Throwable e) { onError(e); return; } } this.rowCountInLastMessage = rowsProcessed; } RowMerger(StreamObserver<FlatRow> observer); static List<FlatRow> toRows(Iterable<ReadRowsResponse> responses); void clearRowInProgress(); @Override final void onNext(ReadRowsResponse readRowsResponse); Integer getRowCountInLastMessage(); ByteString getLastCompletedRowKey(); @Override void onCompleted(); @Override void onError(Throwable e);  }