[
https://issues.apache.org/jira/browse/NIFI-4833?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16392985#comment-16392985
]
ASF GitHub Bot commented on NIFI-4833:
--------------------------------------
Github user bbende commented on the issue:
https://github.com/apache/nifi/pull/2478
@bdesert thanks for the updates, there is one error case I think we need to
handle and then we should be good to go...
The case is if an exception happens half-way through handling the results
and gets caught in the try-catch on lines 385-389, we currently only transfer
the original flow file to failure and return, but the handler may have a flow
file it created and was writing results to, and this would need to be removed
from the session.
The reason this isn't caught in the JUnit tests is because currently
MockHBaseClientService lets you set throwException which throws an exception
right at the beginning of scan before the handler has ever been called. If you
want to create a way to test this you could introduce a new boolean like
throwExceptionAfterNumResults and also take in an integer number of results.
For quick testing I hacked a change into the MockHBaseClientService so that
it throws an exception after the first result:
```
@Override
public void scan(String tableName, String startRow, String endRow,
String filterExpression, Long timerangeMin,
Long timerangeMax, Integer limitRows, Boolean isReversed,
Collection<Column> columns, ResultHandler handler)
throws IOException {
// if (throwException) {
// throw new IOException("exception");
// }
// pass all the staged data to the handler
int resultCount = 0;
for (final Map.Entry<String,ResultCell[]> entry :
results.entrySet()) {
handler.handle(entry.getKey().getBytes(StandardCharsets.UTF_8),
entry.getValue());
resultCount++;
if (resultCount > 0) {
throw new IOException("exception");
}
}
// delegate to the handler
numScans++;
}
```
Then updated the test case:
```
@Test
public void testScanWhenScanThrowsException() {
//hBaseClientService.setThrowException(true);
final Map<String, String> cells = new HashMap<>();
cells.put("cq1", "val1");
cells.put("cq2", "val2");
final long ts1 = 123456789;
hBaseClientService.addResult("row1", cells, ts1);
hBaseClientService.addResult("row2", cells, ts1);
runner.setProperty(ScanHBase.TABLE_NAME, "table1");
runner.setProperty(ScanHBase.START_ROW, "row1");
runner.setProperty(ScanHBase.END_ROW, "row1");
runner.enqueue("trigger flow file");
runner.run();
runner.assertTransferCount(ScanHBase.REL_FAILURE, 1);
runner.assertTransferCount(ScanHBase.REL_SUCCESS, 0);
runner.assertTransferCount(ScanHBase.REL_ORIGINAL, 0);
Assert.assertEquals(0, hBaseClientService.getNumScans());
}
```
> NIFI-4833 Add ScanHBase processor
> ---------------------------------
>
> Key: NIFI-4833
> URL: https://issues.apache.org/jira/browse/NIFI-4833
> Project: Apache NiFi
> Issue Type: New Feature
> Components: Extensions
> Reporter: Ed Berezitsky
> Assignee: Ed Berezitsky
> Priority: Major
>
> Add ScanHBase (new) processor to retrieve records from HBase tables.
> Today there are GetHBase and FetchHBaseRow. GetHBase can pull entire table or
> only new rows after processor started; it also must be scheduled and doesn't
> support incoming . FetchHBaseRow can pull rows with known rowkeys only.
> This processor could provide functionality similar to what could be reached
> by using hbase shell, defining following properties:
> -scan based on range of row key IDsÂ
> -scan based on range of time stamps
> -limit number of records pulled
> -use filters
> -reverse rows
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
