More logs

partition_table.go

@@ -445,19 +445,51 @@ func (p *PartitionTable) loadMessages(ctx context.Context, cons sarama.Partition
  select {
  case err, ok := <-errors:
  if !ok {
+ p.log.Debugf(
+ "load messages exit with err !ok = topic=%s, partition=%v, key=%s, offset=%v, timestamp=%v",
+ msg.Topic,
+ msg.Partition,
+ string(msg.Key),
+ msg.Offset,
+ msg.Timestamp,
+ )
  return nil
  }
  if err != nil {
+ p.log.Debugf(
+ "load messages exit with err = topic=%s, partition=%v, key=%s, offset=%v, timestamp=%v",
+ msg.Topic,
+ msg.Partition,
+ string(msg.Key),
+ msg.Offset,
+ msg.Timestamp,
+ )
  return err
  }
  case msg, ok := <-messages:
  if !ok {
+ p.log.Debugf(
+ "load messages exit with msg !ok = topic=%s, partition=%v, key=%s, offset=%v, timestamp=%v",
+ msg.Topic,
+ msg.Partition,
+ string(msg.Key),
+ msg.Offset,
+ msg.Timestamp,
+ )
  return nil
  }
 
  // This case is for the Tester to achieve synchronity.
  // Nil messages are never generated by the Sarama Consumer
  if msg == nil {
+ p.log.Debugf(
+ "load messages continue msg == nil = topic=%s, partition=%v, key=%s, offset=%v, timestamp=%v",
+ msg.Topic,
+ msg.Partition,
+ string(msg.Key),
+ msg.Offset,
+ msg.Timestamp,
+ )
  continue
  }