virajjasani commented on a change in pull request #2127:
URL: https://github.com/apache/hbase/pull/2127#discussion_r459606308
##########
File path:
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSink.java
##########
@@ -403,13 +398,24 @@ public void stopReplicationSinkServices() {
* Do the changes and handle the pool
* @param tableName table to insert into
* @param allRows list of actions
+ * @param batchRowSizeThreshold rowSize threshold for batch mutation
*/
- private void batch(TableName tableName, Collection<List<Row>> allRows)
throws IOException {
+ private void batch(TableName tableName, Collection<List<Row>> allRows, int
batchRowSizeThreshold)
+ throws IOException {
if (allRows.isEmpty()) {
return;
}
AsyncTable<?> table = getConnection().getTable(tableName);
- List<Future<?>> futures =
allRows.stream().map(table::batchAll).collect(Collectors.toList());
+ List<Future<?>> futures = new ArrayList<>();
+ for (List<Row> rows : allRows) {
+ List<List<Row>> batchRows;
+ if (rows.size() > batchRowSizeThreshold) {
+ batchRows = Lists.partition(rows, batchRowSizeThreshold);
+ } else {
+ batchRows = Collections.singletonList(rows);
+ }
+
futures.addAll(batchRows.stream().map(table::batchAll).collect(Collectors.toList()));
+ }
Review comment:
That's true, they are handled in next batch.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
