Большой расход памяти при обработки Runnable тасков

@Component
public class TestClass extends Verify {
 
    private final ThreadPoolTaskExecutor processExecutor;
    private final BlockingQueue<KPIData> kpiData;
 
    @Autowired
    public TestClass(
            final DbAddresser dbAddresser,
            final MyDao myDao,
            final ThreadPoolTaskExecutor processExecutor) {
        super(VerificationType.KPI, dbAddresser, myDao);
        this.processExecutor = processExecutor;
        kpiData = new ArrayBlockingQueue<>(20);
    }
 
    @Override
    public void verification() throws Exception {
        startedVerification();
        verifyIndicator.setTitle("tasks");
        final Thread consumerKPI = new Thread(() -> {
            ProcessIdGenerator.generate();
            LOG.debug("Consumer thread started....");
            try {
                KPIData kpiData;
                while (!(kpiData = this.kpiData.take()).isLastElement()) {
                    final InconsistsKPICallable callable = new InconsistsKPICallable(((KPIDataImpl) kpiData).getRes());
                    execDispatcher.put(processExecutor.submit(callable));
                    verifyIndicator.incMaxStep();
                }
            } catch (Exception e) {
                LOG.error("Consumer thread error: ", e);
            }
            LOG.debug("Consumer thread finished....");
        });
        consumerKPI.start();
 
        try {
 
            smbpDbAddresser.getKpiAll(kpiData, 5000);
 
            final Pair<Long, Long> minMaxEntities = dbAddresser::getMinMaxEntities;
            if (minMaxEntities != null) {
                long currentEntityId = minMaxEntities.getLeft();
 
                while (currentEntityId <= minMaxEntities.getRight()) {
                    if (stopped) {
                        break;
                    }
 
                    long fromEntityId = currentEntityId;
                    long toEntityId = currentEntityId + 4999;
 
                    incMaxStep();
 
                    InconsistsKPIneedUpdateCallable callable = new InconsistsKPIneedUpdateCallable(fromEntityId, toEntityId);
                    execDispatcher.put(processExecutor.submit(callable));
                    currentEntityId = toEntityId + 1;
                }
            }
        } finally {
            consumerKPI.join();
            waitFinished();
            finishedVerification();
        }
    }
 
    public class InconsistsKPICallable implements Runnable {
        private final List<Pair<Long, String>> kpiList;
 
        InconsistsKPICallable(final List<Pair<Long, String>> kpiList) {
            this.kpiList = kpiList;
        }
 
        @Override
        public void run() {
            try {
                // do hard work with kpiList
            } catch (Exception ex) {
            } finally {
                LOG.debug("InconsistsSmbpKPICallable finished!");
            }
            incCurStep();
        }
    }
 
    public class InconsistsKPIneedUpdateCallable implements Runnable {
        private final Long fromEntityId;
        private final Long toEntityId;
 
        InconsistsKPIneedUpdateCallable(final long fromEntityId, final long toEntityId) {
            this.fromEntityId = fromEntityId;
            this.toEntityId = toEntityId;
        }
 
        @Override
        public void run() {
            try {
                // do hard work with fromEntityId and toEntityId
            } catch (Exception ex) {
            } finally {
                LOG.debug("InconsistsKPIneedUpdateCallable finished for fromEntityId {} toEntityId {}", fromEntityId, toEntityId);
            }
            incCurStep();
        }
    }
}