Manno15 commented on a change in pull request #2296:
URL: https://github.com/apache/accumulo/pull/2296#discussion_r719796657
##########
File path: core/src/main/java/org/apache/accumulo/fate/Fate.java
##########
@@ -228,24 +233,25 @@ public void startTransactionRunners(AccumuloConfiguration
conf) {
Property.MANAGER_FATE_THREADPOOL_SIZE);
fatePoolWatcher = ThreadPools.createGeneralScheduledExecutorService(conf);
fatePoolWatcher.schedule(() -> {
+ // resize the pool if the property changed
ThreadPools.resizePool(pool, conf,
Property.MANAGER_FATE_THREADPOOL_SIZE);
- // Assume a thread could execute up to 100 operations a second.
- // We are sleeping for three seconds. So to calculate max to queue do 3*
100 * numThreads.
- int maxToQueue = 300 *
conf.getCount(Property.MANAGER_FATE_THREADPOOL_SIZE);
- int remaining = maxToQueue - pool.getQueue().size();
- for (int i = 0; i < remaining; i++) {
- try {
- pool.execute(new TransactionRunner());
- } catch (RejectedExecutionException e) {
- // RejectedExecutionException could be shutting down
- if (pool.isShutdown()) {
- // The exception is expected in this case, no need to spam the
logs.
- log.trace("Error adding transaction runner to FaTE executor
pool.", e);
- } else {
- // This is bad, FaTE may no longer work!
- log.error("Error adding transaction runner to FaTE executor
pool.", e);
+ // If the pool grew, then ensure that there is a TransactionRunner for
each thread
+ int needed = conf.getCount(Property.MANAGER_FATE_THREADPOOL_SIZE) -
pool.getQueue().size();
+ if (needed > 0) {
Review comment:
This change looks fine but I wanted to know if there was another reason
you went with this approach instead of the previous one.
--
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]
