Added topic name in kafka consumer log messages
@@ -177,8 +177,8 @@ public class SimpleConsumerWrapper implements PinotStreamConsumer {
void handleConsumerException(Exception e) {
// By default, just log the exception and switch back to CONNECTING_TO_BOOTSTRAP_NODE (which will take care of
// closing the connection if it exists)
- LOGGER.warn("Caught Kafka consumer exception while in state {}, disconnecting and trying again",
- _currentState.getStateValue(), e);
+ LOGGER.warn("Caught Kafka consumer exception while in state {}, disconnecting and trying again for topic {}",
+ _currentState.getStateValue(), _topic, e);
Uninterruptibles.sleepUninterruptibly(250, TimeUnit.MILLISECONDS);
@@ -202,7 +202,7 @@ public class SimpleConsumerWrapper implements PinotStreamConsumer {
try {
_simpleConsumer.close();
} catch (Exception e) {
- LOGGER.warn("Caught exception while closing consumer, ignoring", e);
+ LOGGER.warn("Caught exception while closing consumer for topic {}, ignoring", _topic, e);
}
}
@@ -211,7 +211,7 @@ public class SimpleConsumerWrapper implements PinotStreamConsumer {
_currentPort = _bootstrapPorts[randomHostIndex];
try {
- LOGGER.info("Connecting to bootstrap host {}:{}", _currentHost, _currentPort);
+ LOGGER.info("Connecting to bootstrap host {}:{} for topic {}", _currentHost, _currentPort, _topic);
_simpleConsumer = _simpleConsumerFactory.buildSimpleConsumer(_currentHost, _currentPort, SOCKET_TIMEOUT_MILLIS,
SOCKET_BUFFER_SIZE, _clientId);
setCurrentState(new ConnectedToBootstrapNode());
@@ -270,18 +270,18 @@ public class SimpleConsumerWrapper implements PinotStreamConsumer {
// If we've located a broker
if (_leader != null) {
- LOGGER.info("Located leader broker {}, connecting to it.", _leader);
+ LOGGER.info("Located leader broker {} for topic {}, connecting to it.", _leader, _topic);
setCurrentState(new ConnectingToPartitionLeader());
} else {
// Failed to get the leader broker. There could be a leader election at the moment, so retry after a little
// bit.
- LOGGER.warn("Leader broker is null, retrying leader fetch in 100ms");
+ LOGGER.warn("Leader broker is null for topic {}, retrying leader fetch in 100ms", _topic);
Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
}
} catch (Exception e) {
// Failed to get the leader broker. There could be a leader election at the moment, so retry after a little
// bit.
- LOGGER.warn("Failed to get the leader broker due to exception, retrying in 100ms", e);
+ LOGGER.warn("Failed to get the leader broker for topic {} due to exception, retrying in 100ms", _topic, e);
Uninterruptibles.sleepUninterruptibly(100, TimeUnit.MILLISECONDS);
}
} catch (Exception e) {
@@ -303,7 +303,7 @@ public class SimpleConsumerWrapper implements PinotStreamConsumer {
@Override
void process() {
// If we're already connected to the leader broker, don't disconnect and reconnect
- LOGGER.info("Trying to fetch leader host and port: {}:{}", _leader.host(), _leader.port());
+ LOGGER.info("Trying to fetch leader host and port: {}:{} for topic {}", _leader.host(), _leader.port(), _topic);
if (_leader.host().equals(_currentHost) && _leader.port() == _currentPort) {
setCurrentState(new ConnectedToPartitionLeader());
return;
@@ -356,7 +356,8 @@ public class SimpleConsumerWrapper implements PinotStreamConsumer {
private void setCurrentState(State newState) {
if (_currentState != null) {
- LOGGER.info("Switching from state {} to state {}", _currentState.getStateValue(), newState.getStateValue());
+ LOGGER.info("Switching from state {} to state {} for topic {}",
+ _currentState.getStateValue(), newState.getStateValue(), _topic);
}
_currentState = newState;
