- Integer partition = record.partition();
- String key = record.key() == null ? "NULL" : record.key();
-
- if (!seen.containsKey(partition))
- seen.put(partition, new HashMap<>());
-
- Map<String, Integer> byKey = seen.get(partition);
-
- if (!byKey.containsKey(key))
- byKey.put(key, 0);
-
- int seenByKey = byKey.get(key);
- seenByKey++;
- byKey.put(key, seenByKey);
- }
- }
- }
- catch(WakeupException e)
- {
- log.info("{} - RIIING!", id);
- }
- catch(Exception e)
- {
- log.error("{} - Unexpected error: {}", id, e.toString(), e);
- running.set(false); // Mark the instance as not running
- }
- finally
- {
- log.info("{} - Closing the KafkaConsumer", id);
- consumer.close();
-
- for (Integer partition : seen.keySet())
- {
- Map<String, Integer> byKey = seen.get(partition);
- for (String key : byKey.keySet())
- {
- log.info(
- "{} - Seen {} messages for partition={}|key={}",
- id,
- byKey.get(key),
- partition,
- key);
- }
- }
- seen = null;
-
- log.info("{} - Consumer-Thread exiting", id);
- }
- }