Fix handling of transient kafka errors in the server
@@ -269,8 +269,10 @@ public class LLRealtimeSegmentDataManager extends SegmentDataManager {
highWatermark = messagesAndWatermark.getRight();
} catch (TimeoutException e) {
handleTransientKafkaErrors(e);
+ continue;
} catch (SimpleConsumerWrapper.TransientConsumerException e) {
handleTransientKafkaErrors(e);
+ continue;
} catch (SimpleConsumerWrapper.PermanentConsumerException e) {
segmentLogger.warn("Kafka permanent exception when fetching messages, stopping consumption", e);
throw e;
@@ -278,8 +280,24 @@ public class LLRealtimeSegmentDataManager extends SegmentDataManager {
// Unknown exception from Kafka. Treat as a transient exception.
// One such exception seen so far is java.net.SocketTimeoutException
handleTransientKafkaErrors(e);
+ continue;
}
+ processKafkaEvents(messagesAndOffsets, highWatermark);
+ }
+
+ _serverMetrics.addMeteredTableValue(_metricKeyName, ServerMeter.ROWS_WITH_ERRORS,
+ (long) _fieldExtractor.getTotalErrors());
+ _serverMetrics.addMeteredTableValue(_metricKeyName, ServerMeter.ROWS_NEEDING_CONVERSIONS,
+ (long) _fieldExtractor.getTotalConversions());
+ _serverMetrics.addMeteredTableValue(_metricKeyName, ServerMeter.ROWS_WITH_NULL_VALUES,
+ (long) _fieldExtractor.getTotalNulls());
+ _serverMetrics.addMeteredTableValue(_metricKeyName, ServerMeter.COLUMNS_WITH_NULL_VALUES,
+ (long) _fieldExtractor.getTotalNullCols());
+ return true;
+ }
+
+ private void processKafkaEvents(Iterable<MessageAndOffset> messagesAndOffsets, Long highWatermark) {
Iterator<MessageAndOffset> msgIterator = messagesAndOffsets.iterator();
int indexedMessageCount = 0;
@@ -363,17 +381,6 @@ public class LLRealtimeSegmentDataManager extends SegmentDataManager {
}
}
- _serverMetrics.addMeteredTableValue(_metricKeyName, ServerMeter.ROWS_WITH_ERRORS,
- (long) _fieldExtractor.getTotalErrors());
- _serverMetrics.addMeteredTableValue(_metricKeyName, ServerMeter.ROWS_NEEDING_CONVERSIONS,
- (long) _fieldExtractor.getTotalConversions());
- _serverMetrics.addMeteredTableValue(_metricKeyName, ServerMeter.ROWS_WITH_NULL_VALUES,
- (long) _fieldExtractor.getTotalNulls());
- _serverMetrics.addMeteredTableValue(_metricKeyName, ServerMeter.COLUMNS_WITH_NULL_VALUES,
- (long) _fieldExtractor.getTotalNullCols());
- return true;
- }
-
public class PartitionConsumer implements Runnable {
public void run() {
long initialConsumptionEnd = 0L;
