events buffer is variable, no need to poll for 1000 events, if the bu…

…ffer is 10

kafka/consumer.go

@@ -308,7 +308,7 @@ func NewConsumer(conf *ConfigMap) (*Consumer, error) {
 		c.readerTermChan = make(chan bool)
 
 		/* Start rdkafka consumer queue reader -> events writer goroutine */
-		go consumerReader(c, c.readerTermChan)
+		go consumerReader(c, c.readerTermChan, eventsChanSize)
 	}
 
 	return c, nil
@@ -329,15 +329,20 @@ func (c *Consumer) rebalance(ev Event) bool {
 // consumerReader reads messages and events from the librdkafka consumer queue
 // and posts them on the consumer channel.
 // Runs until termChan closes
-func consumerReader(c *Consumer, termChan chan bool) {
+func consumerReader(c *Consumer, termChan chan bool, eventsChanSize int) {
 
 out:
 	for true {
 		select {
 		case _ = <-termChan:
 			break out
 		default:
-			_, term := c.handle.eventPoll(c.events, 100, 1000, termChan)
+			_, term := c.handle.eventPoll(
+				c.events,
+				100,
+				eventsChanSize - len(c.events), // max events depends on available buffer
+				termChan,
+			)
 			if term {
 				break out
 			}