- }
- // tag::loop[]
- }
- // end::loop[]
- }
- catch(WakeupException e)
- {
- log.info("{} - RIIING!", id);
- }
- catch(Exception e)
- {
- log.error("{} - Unexpected error: {}", id, e.toString());
- }
- finally
- {
- this.lock.lock();
- try
- {
- log.info("{} - Closing the KafkaConsumer", id);
- consumer.close();
- log.info("C - DONE!");
- stopped.signal();
- }
- finally
- {
- this.lock.unlock();
- log.info("{}: Consumed {} messages in total, exiting!", id, consumed);
- }
- }
+
+ recordHandler.calculateSum(topic, partition, offset, key, message);
+
+ consumed++;
+ }
+
+ public void start()
+ {
+ if (running())
+ throw new IllegalStateException("Consumer instance " + id + " is already running!");
+
+ log.info("{} - Starting - consumed {} messages before", id, consumed);
+ errorHandler.clearState();
+ registry.getListenerContainer(id).start();
+ }
+
+ public void stop()
+ {
+ if (!running())
+ throw new IllegalStateException("Consumer instance " + id + " is not running!");
+
+ log.info("{} - Stopping", id);
+ registry.getListenerContainer(id).stop();
+ log.info("{} - Stopped - consumed {} messages so far", id, consumed);