private final String groupId;
private final String id;
private final String topic;
+ private final String autoOffsetReset;
private AtomicBoolean running = new AtomicBoolean();
private long consumed = 0;
String bootstrapServer,
String groupId,
String clientId,
- String topic)
+ String topic,
+ String autoOffsetReset)
{
this.executor = executor;
this.bootstrapServer = bootstrapServer;
this.groupId = groupId;
this.id = clientId;
this.topic = topic;
+ this.autoOffsetReset = autoOffsetReset;
}
@Override
props.put("bootstrap.servers", bootstrapServer);
props.put("group.id", groupId);
props.put("client.id", id);
- props.put("auto.offset.reset", "earliest");
+ props.put("auto.offset.reset", autoOffsetReset);
props.put("key.deserializer", StringDeserializer.class.getName());
props.put("value.deserializer", StringDeserializer.class.getName());
{
boolean stateChanged = running.compareAndSet(false, true);
if (!stateChanged)
- throw new RuntimeException("Consumer instance " + id + " is already running!");
+ throw new IllegalStateException("Consumer instance " + id + " is already running!");
log.info("{} - Starting - consumed {} messages before", id, consumed);
future = executor.submit(this);
{
boolean stateChanged = running.compareAndSet(true, false);
if (!stateChanged)
- throw new RuntimeException("Consumer instance " + id + " is not running!");
+ throw new IllegalStateException("Consumer instance " + id + " is not running!");
log.info("{} - Stopping", id);
consumer.wakeup();
{
log.info("{} - Was already stopped", id);
}
+ catch (Exception e)
+ {
+ log.error("{} - Unexpected exception while trying to stop the consumer", id, e);
+ }
finally
{
log.info("{}: Consumed {} messages in total, exiting!", id, consumed);