diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/scheduler/AbstractSchedulerGroup.java b/pinot-core/src/main/java/org/apache/pinot/core/query/scheduler/AbstractSchedulerGroup.java index 4d4ed89d2585..7cfad7d81af9 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/query/scheduler/AbstractSchedulerGroup.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/query/scheduler/AbstractSchedulerGroup.java @@ -19,7 +19,9 @@ package org.apache.pinot.core.query.scheduler; import com.google.common.base.Preconditions; +import java.util.ArrayList; import java.util.Iterator; +import java.util.List; import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.atomic.AtomicInteger; @@ -64,14 +66,20 @@ public SchedulerQueryContext removeFirst() { } @Override - public void trimExpired(long deadlineMillis) { + public List trimExpired(long deadlineMillis) { + List expired = List.of(); Iterator iter = _pendingQueries.iterator(); while (iter.hasNext()) { SchedulerQueryContext next = iter.next(); if (next.getArrivalTimeMs() < deadlineMillis) { iter.remove(); + if (expired.isEmpty()) { + expired = new ArrayList<>(); + } + expired.add(next); } } + return expired; } @Override diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/scheduler/BinaryWorkloadScheduler.java b/pinot-core/src/main/java/org/apache/pinot/core/query/scheduler/BinaryWorkloadScheduler.java index fdae2557ab57..f818d2827c0d 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/query/scheduler/BinaryWorkloadScheduler.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/query/scheduler/BinaryWorkloadScheduler.java @@ -35,6 +35,7 @@ import org.apache.pinot.core.query.scheduler.resources.QueryExecutorService; import org.apache.pinot.spi.accounting.ThreadAccountant; import org.apache.pinot.spi.env.PinotConfiguration; +import org.apache.pinot.spi.exception.QueryErrorCode; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -72,7 +73,7 @@ public BinaryWorkloadScheduler(PinotConfiguration config, String instanceId, Que super(config, instanceId, queryExecutor, threadAccountant, latestQueryTime, new BinaryWorkloadResourceManager(config)); - _secondaryQueryQ = new SecondaryWorkloadQueue(config, _resourceManager); + _secondaryQueryQ = new SecondaryWorkloadQueue(config, _resourceManager, this::onQueryExpired); _numSecondaryRunners = config.getProperty(MAX_SECONDARY_QUERIES, DEFAULT_MAX_SECONDARY_QUERIES); LOGGER.info("numSecondaryRunners={}", _numSecondaryRunners); _secondaryRunnerSemaphore = new Semaphore(_numSecondaryRunners); @@ -146,6 +147,9 @@ public void run() { try { SchedulerQueryContext request = _secondaryQueryQ.take(); if (request == null) { + // mirrors query completion callback functionality. + _secondaryRunnerSemaphore.release(); + checkStopResourceManager(); continue; } ServerQueryRequest queryRequest = request.getQueryRequest(); @@ -157,6 +161,7 @@ public void run() { public void run() { executorService.releaseWorkers(); schedulerGroup.endQuery(); + _secondaryQueryQ.signalWorkersReleased(); _secondaryRunnerSemaphore.release(); checkStopResourceManager(); } @@ -209,8 +214,27 @@ private void checkStopResourceManager() { synchronized private void failAllPendingQueries() { List pending = _secondaryQueryQ.drain(); for (SchedulerQueryContext queryContext : pending) { - ListenableFuture serverShuttingDown = shuttingDown(queryContext.getQueryRequest()); - queryContext.setResultFuture(serverShuttingDown); + try { + ListenableFuture serverShuttingDown = shuttingDown(queryContext.getQueryRequest()); + queryContext.setResultFuture(serverShuttingDown); + } catch (Throwable t) { + LOGGER.error("Failed to fail pending query: {} on shutdown", queryContext.getQueryRequest().getRequestId(), t); + // Last resort, so one bad query does not leave the remaining ones without a response. + // No-op if the handler already completed the future. + if (queryContext.getResultFuture() != null) { + queryContext.getResultFuture().setException(t); + } + } } } + + private void onQueryExpired(SchedulerQueryContext queryContext) { + ServerQueryRequest queryRequest = queryContext.getQueryRequest(); + long waitMs = System.currentTimeMillis() - queryRequest.getTimerContext().getQueryArrivalTimeMs(); + LOGGER.warn("Dropping secondary query: {} for table: {} after waiting: {}ms in the queue", + queryRequest.getRequestId(), queryRequest.getTableNameWithType(), waitMs); + _serverMetrics.addMeteredTableValue(queryRequest.getTableNameWithType(), ServerMeter.SCHEDULING_TIMEOUT_EXCEPTIONS, + 1L); + queryContext.setResultFuture(immediateErrorResponse(queryRequest, QueryErrorCode.QUERY_SCHEDULING_TIMEOUT)); + } } diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/scheduler/SchedulerGroup.java b/pinot-core/src/main/java/org/apache/pinot/core/query/scheduler/SchedulerGroup.java index a2482ba9978e..b1e0c38f5955 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/query/scheduler/SchedulerGroup.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/query/scheduler/SchedulerGroup.java @@ -18,6 +18,9 @@ */ package org.apache.pinot.core.query.scheduler; +import java.util.List; + + /// Scheduler group is a sub-queue in multi-level scheduling queues. /// This class maintains context information for each of the scheduling /// queues. Each SchedulerGroup is a queue of requests and related accounting. @@ -41,7 +44,8 @@ public interface SchedulerGroup extends SchedulerGroupAccountant { SchedulerQueryContext removeFirst(); /// Remove all the pending queries with arrival time earlier than the deadline - void trimExpired(long deadlineMillis); + /// @return the removed queries, so the caller can complete their result futures. Never null. + List trimExpired(long deadlineMillis); /// @return true if there are no pending queries for this group boolean isEmpty(); diff --git a/pinot-core/src/main/java/org/apache/pinot/core/query/scheduler/SecondaryWorkloadQueue.java b/pinot-core/src/main/java/org/apache/pinot/core/query/scheduler/SecondaryWorkloadQueue.java index 4cd847c2cee6..8a97bdac49fc 100644 --- a/pinot-core/src/main/java/org/apache/pinot/core/query/scheduler/SecondaryWorkloadQueue.java +++ b/pinot-core/src/main/java/org/apache/pinot/core/query/scheduler/SecondaryWorkloadQueue.java @@ -26,6 +26,7 @@ import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; +import java.util.function.Consumer; import javax.annotation.Nullable; import org.apache.pinot.core.query.request.ServerQueryRequest; import org.apache.pinot.core.query.scheduler.fcfs.FCFSSchedulerGroup; @@ -58,10 +59,13 @@ public class SecondaryWorkloadQueue { private final Condition _queryReaderCondition = _queueLock.newCondition(); private final ResourceManager _resourceManager; private final int _queryDeadlineMs; + private final Consumer _expiredQueryHandler; - public SecondaryWorkloadQueue(PinotConfiguration config, ResourceManager resourceManager) { + public SecondaryWorkloadQueue(PinotConfiguration config, ResourceManager resourceManager, + Consumer expiredQueryHandler) { Preconditions.checkNotNull(config); Preconditions.checkNotNull(resourceManager); + Preconditions.checkNotNull(expiredQueryHandler); _queryDeadlineMs = config.getProperty(SECONDARY_QUEUE_QUERY_TIMEOUT, DEFAULT_SECONDARY_QUEUE_QUERY_TIMEOUT_SEC) * 1000; @@ -71,6 +75,7 @@ public SecondaryWorkloadQueue(PinotConfiguration config, ResourceManager resourc _maxPendingPerGroup); _schedulerGroup = new FCFSSchedulerGroup(SECONDARY_WORKLOAD_GROUP_NAME); _resourceManager = resourceManager; + _expiredQueryHandler = expiredQueryHandler; } /// Adds a query to the secondary workload queue. @@ -94,21 +99,52 @@ public void put(SchedulerQueryContext query) /// @return @Nullable public SchedulerQueryContext take() { - _queueLock.lock(); - try { - while (true) { - SchedulerQueryContext schedulerQueryContext; - while ((schedulerQueryContext = takeNextInternal()) == null) { + while (true) { + SchedulerQueryContext schedulerQueryContext = null; + List expired; + _queueLock.lock(); + try { + expired = _schedulerGroup.trimExpired(System.currentTimeMillis() - _queryDeadlineMs); + if (!_schedulerGroup.isEmpty() && _resourceManager.canSchedule(_schedulerGroup)) { + schedulerQueryContext = _schedulerGroup.removeFirst(); + } else if (expired.isEmpty()) { + // Nothing to dispatch and nothing to clean up, so wait for a signal. try { - _queryReaderCondition.await(_wakeUpTimeMs, TimeUnit.MILLISECONDS); + if (_schedulerGroup.isEmpty()) { + _queryReaderCondition.await(); // woken by put() + } else { + _queryReaderCondition.await(_wakeUpTimeMs, TimeUnit.MILLISECONDS); // TTL sweep safety net + } } catch (InterruptedException e) { return null; } } + } finally { + _queueLock.unlock(); + } + + // Complete expired queries outside the lock, since the handler writes to the request channel. + // A failure must not escape: it would drop the query dequeued above and leak the caller's runner permit. + for (SchedulerQueryContext queryContext : expired) { + try { + _expiredQueryHandler.accept(queryContext); + } catch (Throwable t) { + LOGGER.error("Failed to complete expired query: {}", queryContext.getQueryRequest().getRequestId(), t); + // Last resort, so the request always gets a response. No-op if the handler already completed the future. + if (queryContext.getResultFuture() != null) { + queryContext.getResultFuture().setException(t); + } + } + } + if (schedulerQueryContext != null) { + if (LOGGER.isDebugEnabled()) { + ServerQueryRequest queryRequest = schedulerQueryContext.getQueryRequest(); + LOGGER.debug("Scheduling query: {} for group: {} with arrivalTimeMs: {} and numSegments: {}", + queryRequest.getRequestId(), _schedulerGroup.name(), + queryRequest.getTimerContext().getQueryArrivalTimeMs(), queryRequest.getSegmentsToQuery().size()); + } return schedulerQueryContext; } - } finally { - _queueLock.unlock(); } } @@ -125,28 +161,6 @@ public List drain() { return pending; } - private SchedulerQueryContext takeNextInternal() { - long startTimeMs = System.currentTimeMillis(); - long deadlineEpochMillis = startTimeMs - _queryDeadlineMs; - - _schedulerGroup.trimExpired(deadlineEpochMillis); - if (_schedulerGroup.isEmpty() || !_resourceManager.canSchedule(_schedulerGroup)) { - return null; - } - - if (LOGGER.isDebugEnabled()) { - StringBuilder sb = new StringBuilder("SchedulerInfo:"); - sb.append(_schedulerGroup.toString()); - ServerQueryRequest queryRequest = _schedulerGroup.peekFirst().getQueryRequest(); - sb.append(" Group: " + _schedulerGroup.name() + ": [" + queryRequest.getTimerContext().getQueryArrivalTimeMs() - + "," + queryRequest.getRequestId() + "," + queryRequest.getSegmentsToQuery().size() + "," + startTimeMs - + "]"); - LOGGER.debug(sb.toString()); - } - - return _schedulerGroup.removeFirst(); - } - private void checkSchedulerGroupCapacity(SchedulerQueryContext query) throws OutOfCapacityException { if (_schedulerGroup.numPending() >= _maxPendingPerGroup @@ -158,4 +172,14 @@ private void checkSchedulerGroupCapacity(SchedulerQueryContext query) + _resourceManager.getTableThreadsHardLimit()); } } + + /// Signals the reader that reserved threads were released, so canSchedule() may now pass. + public void signalWorkersReleased() { + _queueLock.lock(); + try { + _queryReaderCondition.signal(); + } finally { + _queueLock.unlock(); + } + } } diff --git a/pinot-core/src/test/java/org/apache/pinot/core/query/scheduler/SecondaryWorkloadQueueTest.java b/pinot-core/src/test/java/org/apache/pinot/core/query/scheduler/SecondaryWorkloadQueueTest.java new file mode 100644 index 000000000000..dee31faa76f7 --- /dev/null +++ b/pinot-core/src/test/java/org/apache/pinot/core/query/scheduler/SecondaryWorkloadQueueTest.java @@ -0,0 +1,300 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pinot.core.query.scheduler; + +import com.google.common.util.concurrent.Futures; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.function.Consumer; +import org.apache.pinot.common.metrics.ServerMetrics; +import org.apache.pinot.core.query.scheduler.resources.ResourceManager; +import org.apache.pinot.core.query.scheduler.resources.UnboundedResourceManager; +import org.apache.pinot.spi.env.PinotConfiguration; +import org.apache.pinot.spi.metrics.PinotMetricUtils; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.apache.pinot.core.query.scheduler.TestHelper.createQueryRequest; +import static org.testng.Assert.*; + + +public class SecondaryWorkloadQueueTest { + private static final ServerMetrics METRICS = new ServerMetrics(PinotMetricUtils.getPinotMetricsRegistry()); + private static final String TABLE = "secondaryTable"; + private static final byte[] RESPONSE = new byte[]{1, 2, 3}; + + private static final long AWAIT_MS = 10_000L; + private static final long NEVER_AWAIT_MS = 500L; + private static final long TIMEOUT_MS = 60_000L; + + /// Expires queries after 1s, so a query with an older arrival time is dropped by the TTL sweep. + private static final Map SHORT_DEADLINE = + Map.of(SecondaryWorkloadQueue.SECONDARY_QUEUE_QUERY_TIMEOUT, 1); + /// A wakeup interval far larger than AWAIT_MS, so only an explicit signal can wake a blocked reader in time. + private static final Map NO_POLLING = Map.of(SecondaryWorkloadQueue.QUEUE_WAKEUP_MS, 60_000); + + private final List _resourceManagers = new ArrayList<>(); + private ExecutorService _executor; + + @BeforeMethod + public void beforeMethod() { + _executor = Executors.newCachedThreadPool(); + } + + @AfterMethod + public void afterMethod() { + _executor.shutdownNow(); + _resourceManagers.forEach(ResourceManager::stop); + _resourceManagers.clear(); + } + + // verify that queries are returned in the order they were added, + // and that the expiry handler is not called for live queries. + @Test + public void testTakeReturnsQueriesInArrivalOrder() + throws Exception { + List expiredQueries = new CopyOnWriteArrayList<>(); + SecondaryWorkloadQueue queue = createQueue(Map.of(), expiredQueries::add); + + SchedulerQueryContext first = liveQuery(); + SchedulerQueryContext second = liveQuery(); + queue.put(first); + queue.put(second); + + assertSame(queue.take(), first); + assertSame(queue.take(), second); + assertTrue(expiredQueries.isEmpty()); + } + + // verify that drain() returns all pending queries in arrival order, + // and that the queue is empty afterwards. + @Test + public void testDrainRemovesAllPendingQueries() + throws Exception { + SecondaryWorkloadQueue queue = createQueue(Map.of(), ignored -> { }); + + SchedulerQueryContext first = liveQuery(); + SchedulerQueryContext second = liveQuery(); + queue.put(first); + queue.put(second); + + assertEquals(queue.drain(), List.of(first, second)); + assertTrue(queue.drain().isEmpty()); + } + + /// A query dropped for exceeding the queue deadline must be handed to the expiry handler and have its future + /// completed, otherwise the request never gets a response. Expiring the head must also not stall a live query + /// queued behind it. + @Test + public void testExpiredQueryIsCompletedAndDoesNotBlockLiveQuery() + throws Exception { + List expiredQueries = new CopyOnWriteArrayList<>(); + SecondaryWorkloadQueue queue = createQueue(SHORT_DEADLINE, queryContext -> { + expiredQueries.add(queryContext); + queryContext.setResultFuture(Futures.immediateFuture(RESPONSE)); + }); + + SchedulerQueryContext expired = expiredQuery(); + SchedulerQueryContext live = liveQuery(); + queue.put(expired); + queue.put(live); + + assertSame(queue.take(), live); + assertEquals(expiredQueries, List.of(expired)); + assertSame(expired.getResultFuture().get(AWAIT_MS, MILLISECONDS), RESPONSE); + } + + /// Expiry is not a dispatch event, so a reader with nothing but expired queries keeps waiting. This is the only + /// case that exercises the sweep running with no query to hand back. + @Test(timeOut = TIMEOUT_MS) + public void testExpiredQueryDoesNotWakeReader() + throws Exception { + CountDownLatch handled = new CountDownLatch(1); + SecondaryWorkloadQueue queue = createQueue(SHORT_DEADLINE, queryContext -> handled.countDown()); + + queue.put(expiredQuery()); + Future reader = _executor.submit(queue::take); + + assertTrue(handled.await(AWAIT_MS, MILLISECONDS), "Expired query was never handed to the handler"); + assertThrows(TimeoutException.class, () -> reader.get(NEVER_AWAIT_MS, MILLISECONDS)); + + // The reader is still usable once a live query arrives. + SchedulerQueryContext live = liveQuery(); + queue.put(live); + assertSame(reader.get(AWAIT_MS, MILLISECONDS), live); + } + + /// A failing expiry handler must not break take(): the live query is still returned and the expired query is + /// still completed so its request gets a response. + @Test + public void testExpiryHandlerFailureStillCompletesExpiredQuery() + throws Exception { + SecondaryWorkloadQueue queue = createQueue(SHORT_DEADLINE, queryContext -> { + throw new IllegalStateException("error"); + }); + + SchedulerQueryContext expired = expiredQuery(); + SchedulerQueryContext live = liveQuery(); + queue.put(expired); + queue.put(live); + + assertSame(queue.take(), live); + ExecutionException e = + expectThrows(ExecutionException.class, () -> expired.getResultFuture().get(AWAIT_MS, MILLISECONDS)); + assertTrue(e.getCause() instanceof IllegalStateException); + } + + /// The handler writes to the request channel, so it must not run while the queue lock is held, otherwise Netty + /// threads calling put() would block behind it. + @Test(timeOut = TIMEOUT_MS) + public void testExpiryHandlerRunsWithoutHoldingQueueLock() + throws Exception { + CountDownLatch handlerEntered = new CountDownLatch(1); + CountDownLatch releaseHandler = new CountDownLatch(1); + SecondaryWorkloadQueue queue = createQueue(SHORT_DEADLINE, queryContext -> { + handlerEntered.countDown(); + await(releaseHandler); + }); + + queue.put(expiredQuery()); + Future reader = _executor.submit(queue::take); + assertTrue(handlerEntered.await(AWAIT_MS, MILLISECONDS)); + + // The handler is still running. This blocks forever if take() holds the lock across the callback. + SchedulerQueryContext live = liveQuery(); + queue.put(live); + + releaseHandler.countDown(); + assertSame(reader.get(AWAIT_MS, MILLISECONDS), live); + } + + /// With an empty queue the reader waits untimed, so put() is the only thing that can wake it. + @Test(timeOut = TIMEOUT_MS) + public void testTakeBlocksUntilPutSignals() + throws Exception { + SecondaryWorkloadQueue queue = createQueue(NO_POLLING, ignored -> { }); + + Future reader = _executor.submit(queue::take); + assertThrows(TimeoutException.class, () -> reader.get(NEVER_AWAIT_MS, MILLISECONDS)); + + SchedulerQueryContext live = liveQuery(); + queue.put(live); + assertSame(reader.get(AWAIT_MS, MILLISECONDS), live, "put() did not wake the reader"); + } + + /// A query that cannot be scheduled because the group is at its thread limit must be dispatched as soon as threads + /// are released, without waiting for the TTL sweep. + @Test(timeOut = TIMEOUT_MS) + public void testTakeWakesOnSignalWorkersReleased() + throws Exception { + PinotConfiguration config = new PinotConfiguration(NO_POLLING); + TestResourceManager resourceManager = newResourceManager(config); + resourceManager._canSchedule = false; + SecondaryWorkloadQueue queue = new SecondaryWorkloadQueue(config, resourceManager, ignored -> { }); + + SchedulerQueryContext live = liveQuery(); + queue.put(live); + Future reader = _executor.submit(queue::take); + assertThrows(TimeoutException.class, () -> reader.get(NEVER_AWAIT_MS, MILLISECONDS)); + + resourceManager._canSchedule = true; + queue.signalWorkersReleased(); + assertSame(reader.get(AWAIT_MS, MILLISECONDS), live, "signalWorkersReleased() did not wake the reader"); + } + + /// Admission control ANDs the two conditions, so the pending limit alone never rejects a query. + @Test + public void testOutOfCapacityRequiresPendingAndThreadLimit() + throws Exception { + PinotConfiguration config = new PinotConfiguration(Map.of(SecondaryWorkloadQueue.MAX_PENDING_SECONDARY_QUERIES, 2)); + + TestResourceManager belowThreadLimit = newResourceManager(config); + belowThreadLimit._tableThreadsHardLimit = Integer.MAX_VALUE; + SecondaryWorkloadQueue lenientQueue = new SecondaryWorkloadQueue(config, belowThreadLimit, ignored -> { }); + for (int i = 0; i < 5; i++) { + lenientQueue.put(liveQuery()); + } + + TestResourceManager atThreadLimit = newResourceManager(config); + atThreadLimit._tableThreadsHardLimit = 0; + SecondaryWorkloadQueue strictQueue = new SecondaryWorkloadQueue(config, atThreadLimit, ignored -> { }); + strictQueue.put(liveQuery()); + strictQueue.put(liveQuery()); + assertThrows(OutOfCapacityException.class, () -> strictQueue.put(liveQuery())); + } + + private static SchedulerQueryContext liveQuery() { + return createQueryRequest(TABLE, METRICS); + } + + /// Older than the SHORT_DEADLINE queue deadline, so the TTL sweep drops it. + private static SchedulerQueryContext expiredQuery() { + return createQueryRequest(TABLE, METRICS, System.currentTimeMillis() - TimeUnit.SECONDS.toMillis(10)); + } + + private static void await(CountDownLatch latch) { + try { + latch.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } + + private SecondaryWorkloadQueue createQueue(Map properties, + Consumer expiredQueryHandler) { + PinotConfiguration config = new PinotConfiguration(properties); + return new SecondaryWorkloadQueue(config, newResourceManager(config), expiredQueryHandler); + } + + private TestResourceManager newResourceManager(PinotConfiguration config) { + TestResourceManager resourceManager = new TestResourceManager(config); + _resourceManagers.add(resourceManager); + return resourceManager; + } + + private static class TestResourceManager extends UnboundedResourceManager { + volatile boolean _canSchedule = true; + volatile int _tableThreadsHardLimit = Integer.MAX_VALUE; + + TestResourceManager(PinotConfiguration config) { + super(config); + } + + @Override + public boolean canSchedule(SchedulerGroupAccountant accountant) { + return _canSchedule; + } + + @Override + public int getTableThreadsHardLimit() { + return _tableThreadsHardLimit; + } + } +}