[FEATURE] Support running KoP with Confluent Schema Registry

[BewareMyPower]

Is your feature request related to a problem? Please describe.
Currently Confluent Schema Registry cannot run with KoP because KoP doesn't support CreateTopics and DescribeConfigs requests. Also, it causes that Kafka's AdminClient doesn't work currently.

P.S. since KoP is based on Kafka 2.0.0, the related Confluent Schema Registry version is 5.0.0.

Describe the solution you'd like

• Support CreateTopics: Support CreateTopics request #244
• Support DescribeConfigs: Support DescribeConfigs request #251
• Fix Kafka producer's send callback has no offset: Fill the offset field of PartitionResponse #245
• Fix KafkaAdmin listTopics error: Remove namespace prefix for topics in the default namespace #254

---

The DeleteTopics request is not related to this issue. So we'll do the task later.

[BewareMyPower]

After #244 , the Confluent Schema Registry still failed because it will simply send a record and consume until the offset:

  public void waitUntilKafkaReaderReachesLastOffset(int timeoutMs) throws StoreException {
    long offsetOfLastMessage = getLatestOffset(timeoutMs);
    log.info("Wait to catch up until the offset of the last message at " + offsetOfLastMessage);
    kafkaTopicReader.waitUntilOffset(offsetOfLastMessage, timeoutMs, TimeUnit.MILLISECONDS);
    log.debug("Reached offset at " + offsetOfLastMessage);
  }

      // in getLatestOffset()
      Future<RecordMetadata> ack = producer.send(producerRecord);
      RecordMetadata metadata = ack.get(timeoutMs, TimeUnit.MILLISECONDS);
      this.lastWrittenOffset = metadata.offset();
      log.trace("Noop record's offset is " + this.lastWrittenOffset);
      return this.lastWrittenOffset;

However, KoP doesn't return the offset field to producer now. It's simply set -1. So we need to fix it first.

[BewareMyPower]

Confluent Schema Registry will still fail if _schemas already exist. It's caused by:

      Set<String> allTopics = admin.listTopics().names().get(initTimeout, TimeUnit.MILLISECONDS);
      if (allTopics.contains(topic)) {
        verifySchemaTopic(admin);
      } else {
        createSchemaTopic(admin);
      }

After #206 ,the topic names returned by admin.listTopics()contain the namespace prefix, so it will call createSchemaTopic(admin) again. So we need to fix it and add a unit test of Kafka AdminClient.

[BewareMyPower]

After building KoP with a branch that merges #251 and #254 , the KoP runs well with Confluent Schema Registry no matter if the schema topic exists. So this issue can be closed after they're merged. PTAL, @jiazhai