@Test(expected = ProcessorStateException.class) public void shouldThrowProcessorStateExceptionOnInitializeOffsetsWhenAuthorizationException() throws Exception { final Consumer consumer = mockConsumer(new AuthorizationException("blah")); final AbstractTask task = createTask(consumer); task.updateOffsetLimits(); }
protected void updateOffsetLimits() { log.debug("{} Updating store offset limits {}", logPrefix); for (final TopicPartition partition : partitions) { try { final OffsetAndMetadata metadata = consumer.committed(partition); 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 { protected void updateOffsetLimits() { log.debug("{} Updating store offset limits {}", logPrefix); for (final TopicPartition partition : partitions) { try { final OffsetAndMetadata metadata = consumer.committed(partition); 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 { protected void updateOffsetLimits() { log.debug("{} Updating store offset limits {}", logPrefix); for (final TopicPartition partition : partitions) { try { final OffsetAndMetadata metadata = consumer.committed(partition); 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); }
AbstractTask { protected void updateOffsetLimits() { log.debug("{} Updating store offset limits {}", logPrefix); for (final TopicPartition partition : partitions) { try { final OffsetAndMetadata metadata = consumer.committed(partition); 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); abstract void resume(); abstract void commit(); abstract void suspend(); abstract void close(final boolean clean); final TaskId id(); final String applicationId(); final Set<TopicPartition> partitions(); final ProcessorTopology topology(); final ProcessorContext context(); final ThreadCache cache(); StateStore getStore(final String name); @Override String toString(); String toString(final String indent); }
AbstractTask { protected void updateOffsetLimits() { log.debug("{} Updating store offset limits {}", logPrefix); for (final TopicPartition partition : partitions) { try { final OffsetAndMetadata metadata = consumer.committed(partition); 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); abstract void resume(); abstract void commit(); abstract void suspend(); abstract void close(final boolean clean); final TaskId id(); final String applicationId(); final Set<TopicPartition> partitions(); final ProcessorTopology topology(); final ProcessorContext context(); final ThreadCache cache(); StateStore getStore(final String name); @Override String toString(); String toString(final String indent);  }