Quota Implementation KAFKA-656

Requirements:

There are several quantities we would want to track: 
1. Requests pers second 
2. Bytes written per second 
3. Bytes read per second 

There are two reasonable groupings we would want to aggregate and enforce these thresholds at: 
1. Topic level 
2. Client level (e.g. by client id from the request) 

When a request hits one of these limits we will simply reject it with a QUOTA_EXCEEDED exception. 

Conceptual Design:

This is psuedo-code obviously:

bool checkRequestQuotas('client_ip', 'topic', 'bytes_to_read', 'bytes_to_write')
{
	Meter rps = Metrics.newMeter(class, 'requests-${topic}', 'requests', TimeUnit.SECONDS);
	rps.mark();


	if (rps.getOneMinuteRate() >= quotas.getRPSLimit('topic'))
	{
		return false;
	}
}
bool topicRPS(string topic)
{
	Meter rps = Metrics.newMeter(class, 'requests-${topic}', 'requests', TimeUnit.SECONDS);
	rps.mark();
	if (rps.getOneMinuteRate() >= quotas.getRPSLimit('topic'))	{		return false;	}}