Fix error OutOfMemoryError while using KeyValue<GenericRecord, GenericRecord>

[eolivelli]

Motivation

Currently you cannot consume a topic with schema KeyValue<GenericRecord, GenericRecord> due to a problem in HttpLookupService.
The HttpLookupService download the Schema in JSON format but the KeyValue schema is expected to be encoded in binary form.

The error you see in current pulsar master is:

org.apache.pulsar.client.api.PulsarClientException: java.util.concurrent.ExecutionException: java.lang.OutOfMemoryError: Java heap space

	at org.apache.pulsar.client.api.PulsarClientException.unwrap(PulsarClientException.java:1027)
	at org.apache.pulsar.client.impl.ConsumerBuilderImpl.subscribe(ConsumerBuilderImpl.java:102)
	at org.apache.pulsar.broker.service.KeyValueTest.keyValueAutoConsumeTest(KeyValueTest.java:98)
	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:132)
	at org.testng.internal.TestInvoker.invokeMethod(TestInvoker.java:599)
	at org.testng.internal.TestInvoker.invokeTestMethod(TestInvoker.java:174)
	at org.testng.internal.MethodRunner.runInSequence(MethodRunner.java:46)
	at org.testng.internal.TestInvoker$MethodInvocationAgent.invoke(TestInvoker.java:822)
	at org.testng.internal.TestInvoker.invokeTestMethods(TestInvoker.java:147)
	at org.testng.internal.TestMethodWorker.invokeTestMethods(TestMethodWorker.java:146)
	at org.testng.internal.TestMethodWorker.run(TestMethodWorker.java:128)
	at java.util.ArrayList.forEach(ArrayList.java:1259)
	at org.testng.TestRunner.privateRun(TestRunner.java:764)
	at org.testng.TestRunner.run(TestRunner.java:585)
	at org.testng.SuiteRunner.runTest(SuiteRunner.java:384)
	at org.testng.SuiteRunner.runSequentially(SuiteRunner.java:378)
	at org.testng.SuiteRunner.privateRun(SuiteRunner.java:337)
	at org.testng.SuiteRunner.run(SuiteRunner.java:286)
	at org.testng.SuiteRunnerWorker.runSuite(SuiteRunnerWorker.java:53)
	at org.testng.SuiteRunnerWorker.run(SuiteRunnerWorker.java:96)
	at org.testng.TestNG.runSuitesSequentially(TestNG.java:1218)
	at org.testng.TestNG.runSuitesLocally(TestNG.java:1140)
	at org.testng.TestNG.runSuites(TestNG.java:1069)
	at org.testng.TestNG.run(TestNG.java:1037)
	at com.intellij.rt.testng.IDEARemoteTestNG.run(IDEARemoteTestNG.java:66)
	at com.intellij.rt.testng.RemoteTestNGStarter.main(RemoteTestNGStarter.java:109)
Caused by: java.util.concurrent.ExecutionException: java.lang.OutOfMemoryError: Java heap space
	at java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:357)
	at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1908)
	at org.apache.pulsar.client.impl.ConsumerBuilderImpl.subscribe(ConsumerBuilderImpl.java:100)
	... 28 more
Caused by: java.lang.OutOfMemoryError: Java heap space
	at org.apache.pulsar.common.schema.KeyValue.decode(KeyValue.java:135)
	at org.apache.pulsar.client.impl.schema.KeyValueSchemaInfo.decodeKeyValueSchemaInfo(KeyValueSchemaInfo.java:201)
	at org.apache.pulsar.client.impl.schema.KeyValueSchema.configureSchemaInfo(KeyValueSchema.java:196)
	at org.apache.pulsar.client.impl.PulsarClientImpl.lambda$preProcessSchemaBeforeSubscribe$26(PulsarClientImpl.java:875)
	at org.apache.pulsar.client.impl.PulsarClientImpl$$Lambda$766/571497590.apply(Unknown Source)

Modifications

Use existing utility functions to convert the JSON representation of the KeyValue schema to the expected format.

Verifying this change

This change added tests

[eolivelli]

@sijie thanks for your review