private final Consumer<String, AbstractMessageTo> consumer;
private final ZoneId zoneId;
private final int numShards;
+ private final Duration pollingInterval;
private final int bufferSize;
private final Clock clock;
private final boolean[] isShardOwned;
Consumer<String, AbstractMessageTo> dataChannelConsumer,
ZoneId zoneId,
int numShards,
+ Duration pollingInterval,
int bufferSize,
Clock clock,
InfoChannel infoChannel,
this.producer = producer;
this.zoneId = zoneId;
this.numShards = numShards;
+ this.pollingInterval = pollingInterval;
this.bufferSize = bufferSize;
this.clock = clock;
this.isShardOwned = new boolean[numShards];
"Could not publish instance {} as owner of shard {}: {}",
instanceId,
partition,
- throwable))
+ throwable.toString()))
+ .onErrorComplete()
.block();
});
{
try
{
- ConsumerRecords<String, AbstractMessageTo> records = consumer.poll(Duration.ofMinutes(1));
+ ConsumerRecords<String, AbstractMessageTo> records = consumer.poll(pollingInterval);
log.info("Fetched {} messages", records.count());
if (loadInProgress)