class AbstractTask {

    protected void updateOffsetLimits() {
        log.debug("{} Updating store offset limits {}", logPrefix);
        for (final TopicPartition partition : partitions) {
            try {
                final OffsetAndMetadata metadata = consumer.committed(partition); // TODO: batch API?
                stateMgr.putOffsetLimit(partition, metadata != null ? metadata.offset() : 0L);
            } catch (final AuthorizationException e) {
                throw new ProcessorStateException(String.format("task [%s] AuthorizationException when initializing offsets for %s", id, partition), e);
            } catch (final WakeupException e) {
                throw e;
            } catch (final KafkaException e) {
                throw new ProcessorStateException(String.format("task [%s] Failed to initialize offsets for %s", id, partition), e);
            }
        }
    }

      AbstractTask(final TaskId id,
                 final String applicationId,
                 final Collection<TopicPartition> partitions,
                 final ProcessorTopology topology,
                 final Consumer<byte[], byte[]> consumer,
                 final ChangelogReader changelogReader,
                 final boolean isStandby,
                 final StateDirectory stateDirectory,
                 final ThreadCache cache,
                 final StreamsConfig config);

}