Add support for Kafka 2.4.1

[yonatane]

I need to support adding a 2.4.1 cluster. I've followed the changes made in the previous patch version upgrade from 2.1.0 to 2.1.1 and made similar changes to add 2.4.1 and in some places replace 2.4.0 with 2.4.1.

I ran it locally and added a real 2.4.1 cluster as a sanity check.

One unit test failed though, but that exact test failed without my changes so I leave the test log here for the experts:

2020-07-12 13:05:25,885 - [ERROR] o.a.z.s.NIOServerCnxnFactory - Thread Thread[Thread-33,5,main] died
java.lang.InterruptedException: sleep interrupted
	at java.base/java.lang.Thread.sleep(Native Method)
	at kafka.manager.TestKafkaManager$$anon$3.$anonfun$run$8(TestKafkaManager.scala:92)
	at kafka.manager.TestKafkaManager$$anon$3.$anonfun$run$8$adapted(TestKafkaManager.scala:89)
  | => rat scala.Option.foreach(Option.scala:407)
	at kafka.manager.TestKafkaManager$$anon$3.run(TestKafkaManager.scala:89)
2020-07-12 13:05:25,885 - [ERROR] o.a.z.s.NIOServerCnxnFactory - Thread Thread[Thread-31,5,main] died
java.lang.InterruptedException: sleep interrupted
	at java.base/java.lang.Thread.sleep(Native Method)
	at kafka.manager.TestKafkaManager$$anon$1.run(TestKafkaManager.scala:67)
2020-07-12 13:05:25,888 - [ERROR] o.a.z.s.NIOServerCnxnFactory - Thread Thread[Thread-32,5,main] died
org.apache.kafka.common.errors.InterruptException: java.lang.InterruptedException
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeThrowInterruptException(ConsumerNetworkClient.java:517)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:278)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
	at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1300)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1240)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1211)
	at kafka.test.NewKafkaManagedConsumer.read(SeededBroker.scala:248)
	at kafka.manager.TestKafkaManager$$anon$2.$anonfun$run$5(TestKafkaManager.scala:76)
	at kafka.manager.TestKafkaManager$$anon$2.$anonfun$run$5$adapted(TestKafkaManager.scala:76)
	at scala.Option.map(Option.scala:230)
Caused by: java.lang.InterruptedException: null
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.maybeThrowInterruptException(ConsumerNetworkClient.java:517)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:278)
	at org.apache.kafka.clients.consumer.internals.ConsumerNetworkClient.poll(ConsumerNetworkClient.java:233)
	at org.apache.kafka.clients.consumer.KafkaConsumer.pollForFetches(KafkaConsumer.java:1300)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1240)
	at org.apache.kafka.clients.consumer.KafkaConsumer.poll(KafkaConsumer.java:1211)
	at kafka.test.NewKafkaManagedConsumer.read(SeededBroker.scala:248)
	at kafka.manager.TestKafkaManager$$anon$2.$anonfun$run$5(TestKafkaManager.scala:76)
	at kafka.manager.TestKafkaManager$$anon$2.$anonfun$run$5$adapted(TestKafkaManager.scala:76)
	at scala.Option.map(Option.scala:230)

I confirm that this contribution is made under the terms of the license found in the root directory of this repository's source tree and that I have the authority necessary to make this contribution on behalf of its copyright owner.