make HighLevelConsumer poll for records without waiting
Closes from arjun4084346/consumerImprovement
@@ -107,6 +107,7 @@ public abstract class HighLevelConsumer<K,V> extends AbstractIdleService {
private final int offsetsCommitNumRecordsThreshold;
private final int offsetsCommitTimeThresholdSecs;
private long lastCommitTime = System.currentTimeMillis();
+ protected volatile boolean shutdownRequested = false;
private static final Config FALLBACK =
ConfigFactory.parseMap(ImmutableMap.<String, Object>builder()
@@ -211,12 +212,11 @@ public abstract class HighLevelConsumer<K,V> extends AbstractIdleService {
// Method that starts threads that processes queues
processQueues();
// Main thread that constantly polls messages from kafka
- consumerExecutor.scheduleAtFixedRate(new Runnable() {
- @Override
- public void run() {
+ consumerExecutor.execute(() -> {
+ while (!shutdownRequested) {
consume();
}
- }, 0, 50, TimeUnit.MILLISECONDS);
+ });
}
/**
@@ -277,6 +277,7 @@ public abstract class HighLevelConsumer<K,V> extends AbstractIdleService {
@Override
public void shutDown() {
+ shutdownRequested = true;
ExecutorsUtils.shutdownExecutorService(this.consumerExecutor, Optional.of(log), 5000, TimeUnit.MILLISECONDS);
ExecutorsUtils.shutdownExecutorService(this.queueExecutor, Optional.of(log), 5000, TimeUnit.MILLISECONDS);
try {
