class SaveOffsetsOnRebalance[F[_]: Applicative] extends RebalanceListener1[F] {
  import RebalanceCallback._
  def onPartitionsAssigned(partitions: Nes[TopicPartition]) =
    for {
      // read the offsets from an external store using some custom code not described here
      offsets <- lift(readOffsetsFromExternalStore[F](partitions))
      a       <- offsets.toList.foldMapM { case (partition, offset) => seek(partition, offset) }
    } yield a
  def onPartitionsRevoked(partitions: Nes[TopicPartition]) =
    for {
      positions <- partitions.foldM(Map.empty[TopicPartition, Offset]) {
        case (offsets, partition) =>
          for {
            position <- position(partition)
          } yield offsets + (partition -> position)
      }
      // save the offsets in an external store using some custom code not described here
      a <- lift(saveOffsetsInExternalStore[F](positions))
    } yield a
  // do not need to save the offsets since these partitions are probably owned by other consumers already
  def onPartitionsLost(partitions: Nes[TopicPartition]) = empty
}