tkhurana commented on code in PR #2297:
URL: https://github.com/apache/phoenix/pull/2297#discussion_r2415149393
##########
phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/write/AbstractParallelWriterIndexCommitter.java:
##########
@@ -128,7 +149,6 @@ public void write(Multimap<HTableInterfaceReference,
Mutation> toWrite,
*/
Set<Entry<HTableInterfaceReference, Collection<Mutation>>> entries =
toWrite.asMap().entrySet();
- tasks = new TaskBatch<Void>(entries.size());
Review Comment:
Another option could have been to create a new local tasks object and
introduce a new abstract API on the AbstractParallelWriterIndexCommiter
submitTasks(tasks) which the two derived classes ParallelWriter and
LazyParallelWriter can implement since that is where there is a difference.
##########
phoenix-core-server/src/main/java/org/apache/phoenix/hbase/index/write/LazyParallelWriterIndexCommitter.java:
##########
@@ -17,17 +17,40 @@
*/
package org.apache.phoenix.hbase.index.write;
+import org.apache.hadoop.hbase.client.Mutation;
+import
org.apache.phoenix.hbase.index.exception.SingleIndexWriteFailureException;
+import org.apache.phoenix.hbase.index.parallel.TaskBatch;
+import org.apache.phoenix.hbase.index.table.HTableInterfaceReference;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.phoenix.thirdparty.com.google.common.collect.Multimap;
+
/**
* Like the {@link ParallelWriterIndexCommitter}, but does not block
*/
public class LazyParallelWriterIndexCommitter extends
AbstractParallelWriterIndexCommitter {
- // for testing
- public LazyParallelWriterIndexCommitter(String hbaseVersion) {
- super(hbaseVersion);
- }
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(LazyParallelWriterIndexCommitter.class);
public LazyParallelWriterIndexCommitter() {
super();
}
+
+ @Override
+ public void write(Multimap<HTableInterfaceReference, Mutation> toWrite,
+ final boolean allowLocalUpdates, final int clientVersion)
+ throws SingleIndexWriteFailureException {
+
+ TaskBatch<Void> tasks = new TaskBatch<>(toWrite.asMap().size());
+ super.addTasks(toWrite, allowLocalUpdates, clientVersion, tasks);
+ try {
+ pool.submitOnly(tasks);
+ } catch (Exception e) {
+ LOGGER.error("Error while submitting the task.", e);
+ propagateFailure(e.getCause());
+ }
Review Comment:
Basically move this logic into an abstract submitTasks(tasks) method with
the derived classes implementing it differently.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
