Added error handling and comments

Author: brianhks

src/main/java/org/kairosdb/kafka/monitor/OffsetListenerService.java

@@ -218,14 +218,34 @@ public void execute(JobExecutionContext jobExecutionContext) throws JobExecution
 
 			for (GroupStats groupStats : m_offsetsTracker.copyOfCurrentStats())
 			{
-				Map<Integer, Long> latestOffsets = getLatestTopicOffsets(groupStats.getTopic());
+				Map<Integer, Long> latestOffsets = null;
+
+				try
+				{
+					latestOffsets = getLatestTopicOffsets(groupStats.getTopic());
+				}
+				catch (Exception e)
+				{
+					logger.error("Error reading kafka offsets for topic: "+groupStats.getTopic(), e);
+
+					//Report failure metric
+					ImmutableSortedMap<String, String> offsetTags = new ImmutableSortedMap.Builder<String, String>(Ordering.natural())
+							.putAll(m_monitorConfig.getAdditionalTags())
+							.put("host", m_clientId).build();
+
+					DataPointEvent failureEvent = new DataPointEvent(m_monitorConfig.getGatherFailureMetric(),
+							offsetTags,
+							m_dataPointFactory.createDataPoint(now, 1));
+					postEvent(failureEvent);
+				}
 
 				ImmutableSortedMap<String, String> groupTags = new ImmutableSortedMap.Builder<String, String>(Ordering.natural())
 						.putAll(m_monitorConfig.getAdditionalTags())
 						.put("group", groupStats.getGroupName())
 						.put("proxy_group", groupStats.getProxyGroup())
 						.put("topic", groupStats.getTopic()).build();
 
+				//Report consumer rate
 				DataPointEvent event = new DataPointEvent(m_monitorConfig.getConsumerRateMetric(),
 						groupTags,
 						m_dataPointFactory.createDataPoint(now, groupStats.getConsumeCount()));
@@ -244,30 +264,34 @@ public void execute(JobExecutionContext jobExecutionContext) throws JobExecution
 							.put("topic", groupStats.getTopic())
 							.put("partition", String.valueOf(offsetStat.getPartition())).build();
 
+					//Report offset age
 					DataPointEvent offsetAge = new DataPointEvent(m_monitorConfig.getOffsetAgeMetric(),
 							partitionTags,
 							m_dataPointFactory.createDataPoint(now, (now - offsetStat.getCommitTime())));
 					postEvent(offsetAge);
 
-					Long latestOffset = latestOffsets.get(offsetStat.getPartition());
+					Long latestOffset = latestOffsets != null ? latestOffsets.get(offsetStat.getPartition()) : null;
 					if (latestOffset != null) //in case something goes bananas
 					{
 						long partitionLag = calculateDiff(latestOffset, offsetStat.getOffset());
 						groupLag += partitionLag;
 
+						//Report partition lag
 						DataPointEvent partitionLagEvent = new DataPointEvent(m_monitorConfig.getPartitionLagMetric(),
 								partitionTags,
 								m_dataPointFactory.createDataPoint(now, partitionLag));
 						postEvent(partitionLagEvent);
 					}
 				}
 
+				//Report group lag
 				DataPointEvent groupLagEvent = new DataPointEvent(m_monitorConfig.getGroupLagMetric(),
 						groupTags,
 						m_dataPointFactory.createDataPoint(now, groupLag));
 				postEvent(groupLagEvent);
 
 
+				//Report time to process lag
 				long secToProcess = 0;
 				if (groupStats.getCurrentRate() != 0.0)
 					secToProcess = (long)((double)groupLag / groupStats.getCurrentRate());

src/main/java/org/kairosdb/kafka/monitor/PartitionedOffsetReader.java

@@ -86,6 +86,7 @@ protected void readTopic()
 			ownedTopics.add(record.key());
 		}
 
+		//todo: maybe not send this every time
 		for (String ownedTopic : ownedTopics)
 		{
 			m_producer.send(new ProducerRecord<String, String>(