Added a new MetadataUtils with an implementation of createKafkaMetadataIfMissing…

[wmccarley]

…that does not overwrite the cluster metadata. Modified several use cases to support the change.

Logic for the createKafkaMetadataIfMissing implementation is as follows:

• If the tenant does not exist it will be created
• If the tenant exists but the allowedClusters list does not include the cluster this method will add the cluster to the allowedClusters list
• If the namespace does not exist it will be created
• If the namespace exists but the replicationClusters list does not include the cluster this method will add the cluster to the replicationClusters list
• If the offset topic does not exist it will be created
• If the offset topic exists but some partitions are missing, the missing partitions will be created

[wmccarley]

node-rdkafka fails KafkaIntegrationTest on my machine with: java.lang.IllegalArgumentException: Invalid version for API key API_VERSIONS: 3

I suspect this is also where it is failing in github actions but not sure because I can't see the logs. Any help fixing this would be nice, I can't figure out how this change could have introduced this bug.......

[jiazhai]

@wmccarley from the log(https://github.com/streamnative/kop/pull/168/checks?check_run_id=942014543), it looks like a container issue, I will re-trigger it.

[ERROR] simpleProduceAndConsume(io.streamnative.pulsar.handlers.kop.KafkaIntegrationTest)  Time elapsed: 78.686 s  <<< FAILURE!
org.testcontainers.containers.ContainerLaunchException: Container startup failed
	at org.testcontainers.containers.GenericContainer.doStart(GenericContainer.java:322)
	at org.testcontainers.containers.GenericContainer.start(GenericContainer.java:302)
	at io.streamnative.pulsar.handlers.kop.KafkaIntegrationTest.simpleProduceAndConsume(KafkaIntegrationTest.java:192)
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
	at java.lang.reflect.Method.invoke(Method.java:498)
	at org.testng.internal.MethodInvocationHelper.invokeMethod(MethodInvocationHelper.java:124)
	at org.testng.internal.InvokeMethodRunnable.runOne(InvokeMethodRunnable.java:54)
	at org.testng.internal.InvokeMethodRunnable.run(InvokeMethodRunnable.java:44)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)
Caused by: org.rnorth.ducttape.RetryCountExceededException: Retry limit hit with exception
	at org.rnorth.ducttape.unreliables.Unreliables.retryUntilSuccess(Unreliables.java:88)
	at org.testcontainers.containers.GenericContainer.doStart(GenericContainer.java:315)