purgatory就是炼狱的意思。
当一个DelayedOperation需要被delay时，它就被放到DelayedOperationPurgatory，相当于进行一个等待池。上一篇blog提到过，DelayedOperation想要摆脱delay状态，需要由事件来触发对它状态的检查，或者是超时时间到了。
这个逻辑看起来挺简单，但高效实现却挺复杂。0.9.0版本的Kafka重新设计了这个purgatory(相比0.8版本)，设计思路在Kafka的这篇文档Purgatory Redesign Proposa。我翻译并注释了一下，放在Kafka之Purgatory Redesign Proposal （翻译）
purgatory的实现使用了两个缓存，这里先讲第一个。
直观的，我们需要一个key绑定到DelayedOperation上，来说明这个DelayedOperation会由哪些事件触发，而且一个DelayedOperation可以绑定到多个key, 一个key也可能跟多个DelayedOperation有关。所以这是一个多对多的映射，一边是事件，一边是DelayedOperation。
DelayedOperation pool
DelayedOperationPurgatory就提供了这种功能。它把这种映射保存到一个Pool里。这个Pool实际上是对一个ConcurrentHashMap的封装.
private val watchersForKey = new Pool[Any, Watchers](Some((key: Any) => new Watchers(key)))
Pool构造器的参数是一个value factory，即当这个pool里一个key没有value时，就用这个函数生成一个value。所以，对于value为空的key，这个pool就会构造一个watch这个key的Watchers.
Watchers
这个Pool的key是Any类型，也就是Scala里所有对象的基类，value是Watchers。注释里这么描述Watchers的
A linked list of watched delayed operations based on some key
Watchers把这些DelayedOperation保存在自己的一个instance field里
private[this] val operations = new LinkedList[T]()
由于Watchers是DelayedOperationPurgatory的内部类，T就源于DelayedOperationPurgatory的签名
class DelayedOperationPurgatory[T <: DelayedOperation](purgatoryName: String, brokerId: Int = 0, purgeInterval: Int = 1000)
extends Logging with KafkaMetricsGroup
所以operations这个域里保存的就是一个DelayedOperation的列表。
所以，DelayedOperationPurgatory用来保存DelayedOperation的数据结构是一个Map[Any, List[DelayedOpertion]]的结构。
对于Watchers来说，有两件事要做
- 当它关注的key有事件发生时，需要调用它的方法来遍历operations，找出其中可以被complete（即不再被delay）的operation。这个方法就是tryCompleteWatched
- 由于一个DelayedOperation可以对应多个key，所以当这个Watchers对应的key没有被触发，它保存的operations里的元素仍然可能由于其它的key触发而而被complete。所以外界需要能主动地检测这个Watchers里的哪些operation已经被complete了，并且移除这些元素。这个方法就是purgeCompleted。
做这两件事情的实机很重要，下面来分析一下
- 第一件事，在产生事件的地方进行检测就好。比如fetch线程处理fetch请求的过程，以及produce request的处理过程中，可以调用tryCompleteWatched。
- 第二件事的处理实机比较不好确定。因为当把一个request从某个key的watchers中移除以后，它可能还在另一个key的watchers里。而每次移除一个request,都要调用purgeCompleted显然不现实。但是0.9.0的实现中引用了新的数据结构来对request的超时进行检测，通过它可以准确获得某个时刻在purgatory中的请求数量(但并不是server中的DelayedOperation的数量，因为超时的DelayedOperation会被放入一个线程池执行它的回调，所以总的数量还需要加上线程池中的Operation数量, 而且这个线程池是一个FixedThreadPool，它使用一个*的queue)。具体的作法请参照Kafka之Purgatory Redesign Proposal （翻译）