Page History

/**
 * This is the basic interface we need to implement in the Kafka broker. This class will record all events processed by the broker. 
 */
public interface QuotaManager {
    /** 
     * The function check should be called before processing any request. If this returns false, then some Quota action has to be taken.
     */
    <T extends RequestOrResponse> boolean check(T request);

    /**
     * This function is called after any client request has been processed and lets the implementation track information about the response. 
     * This is useful for consumer side quotas where we may not know the response sizes and number of messages before processing the request entirely.
     *
     * T : T can be any type of request or response. Typically (FetchRequest or ProducerRequest)
     * C : C is basically the response object. Typically (FetchResponse or ProducerResponse).
     *
     */
	<T extends RequestOrResponse, C extends RequestOrResponse> void onResponse(T request, C response);

	void shutdown();
}

// Example for consumers
def handleFetchRequest() {
  val fetchRequest = request.requestObj.asInstanceOf[FetchRequest]
  if(!quotaManager.check(fetchRequest))
    // take quota actions

  // Notify the quota manager after the request has been processed
  def sendResponseCallback(responsePartitionData: Map[TopicAndPartition, FetchResponsePartitionData]) {
      val fetchResponse = FetchResponse(fetchRequest.correlationId, responsePartitionData)
     quotaManager.notify(fetchRequest, fetchResponse);
  }
}

// Example for producers

  def handleProducerRequest(request: RequestChannel.Request) {
    val produceRequest = request.requestObj.asInstanceOf[ProducerRequest]

  if(!quotaManager .check(fetchRequest))
    // take quota actions

    // the callback for sending a produce response
    def sendResponseCallback(responseStatus: Map[TopicAndPartition, ProducerResponseStatus]) {
      var errorInResponse = false
      responseStatus.foreach { case (topicAndPartition, status) =>
        // we only print warnings for known errors here; if it is unknown, it will cause
        // an error message in the replica manager
        if (status.error != ErrorMapping.NoError && status.error != ErrorMapping.UnknownCode) {
          debug("Produce request with correlation id %d from client %s on partition %s failed due to %s"
            .format(produceRequest.correlationId, produceRequest.clientId,
            topicAndPartition, ErrorMapping.exceptionNameFor(status.error)))
          errorInResponse = true
        }
      }
      if (produceRequest.requiredAcks == 0) {
        // no operation needed if producer request.required.acks = 0; however, if there is any error in handling
        // the request, since no response is expected by the producer, the server will close socket server so that
        // the producer client will know that some error has happened and will refresh its metadata
        if (errorInResponse) {
          info("Close connection due to error handling produce request with correlation id %d from client id %s with ack=0"
                  .format(produceRequest.correlationId, produceRequest.clientId))
          requestChannel.closeConnection(request.processor, request)
        } else {
          requestChannel.noOperation(request.processor, request)
        }
      } else {
        val response = ProducerResponse(produceRequest.correlationId, responseStatus)
        requestChannel.sendResponse(new RequestChannel.Response(request, new BoundedByteBufferSend(response)))
      }
    }
    // only allow appending to internal topic partitions
    // if the client is not from admin
    val internalTopicsAllowed = produceRequest.clientId == AdminUtils.AdminClientId
    // call the replica manager to append messages to the replicas
    replicaManager.appendMessages(
      produceRequest.ackTimeoutMs.toLong,
      produceRequest.requiredAcks,
      internalTopicsAllowed,
      produceRequest.data,
      sendResponseCallback)
    // if the request is put into the purgatory, it will have a held reference
    // and hence cannot be garbage collected; hence we clear its data here in
    // order to let GC re-claim its memory since it is already appended to log
    produceRequest.emptyData()
  }