title: Kafka Server not starting up. Gives: Fatal error during KafkaServer startup. Prepare to shutdown: java.lang.NullPointerException
body: <p>I am trying to run Junit tests for my application, and trying to start the Kafka services as part of my tests.
  But, I am getting the <code>NullPointerException</code> after the zookeeper is started and the Kafka server ends abruptly.</p>
  <blockquote>
  <p>Kafka Client Version: 2.4.1</p>
  </blockquote>
  <pre><code>Properties props = new Properties();
      props.setProperty(KafkaConfig.HostNameProp(), &quot;localhost&quot;);
      props.setProperty(KafkaConfig.PortProp(), String.valueOf(KAFKA_BROKER_PORT));
      props.setProperty(KafkaConfig.BrokerIdProp(), String.valueOf(KAFKA_BROKER_ID));
      props.setProperty(KafkaConfig.LogDirProp(), Environment.getLogsDir());
      props.setProperty(KafkaConfig.LogDirProp(), System.getProperty(&quot;java.io.tmpdir&quot;) + &quot;/kafka-logs&quot;);
      props.setProperty(&quot;enable.zookeeper&quot;, &quot;false&quot;);
      props.setProperty(&quot;zookeeper.connect&quot;, &quot;localhost:2181&quot;);
      props.setProperty(&quot;zookeeper.connection.timeout.ms&quot;, &quot;10000&quot;);
      props.setProperty(&quot;offsets.topic.replication.factor&quot;, &quot;1&quot;);
      props.setProperty(&quot;log.flush.interval.messages&quot;, &quot;1&quot;);
      props.setProperty(KafkaConfig.GroupMaxSessionTimeoutMsProp(), &quot;100000&quot;);
  
      // flush every message.
      props.setProperty(&quot;log.flush.interval&quot;, &quot;1&quot;);
  
      // flush every 1ms
      props.setProperty(&quot;log.default.flush.scheduler.interval.ms&quot;, &quot;1&quot;);
      props.setProperty(&quot;log.flush.scheduler.interval.ms&quot;, &quot;1&quot;);
  
  
      kafkaServerStartable = KafkaServerStartable.fromProps(props);
  
  
      Runtime.getRuntime().addShutdownHook(new Thread() {
          @Override
          public void run() {
              if (kafkaServerStartable != null) {
                  kafkaServerStartable.shutdown();
              }
          };
  
        });
      System.setProperty(&quot;zookeeper.sasl.clientconfig&quot;, &quot;test&quot;);
      System.setProperty(&quot;zookeeper.sasl.client&quot;, &quot;false&quot;);
      kafkaServerStartable.startup();
  </code></pre>
  <p>Any help/leads will be helpful. Thanks!</p>
  <p>StackTrace/Logs:</p>
  <pre><code>2021-05-12 11:05:29,944 INFO [kafka.server.KafkaServer][main] [] starting
  2021-05-12 11:05:29,948 INFO [kafka.server.KafkaServer][main] [] Connecting to zookeeper on localhost:2181
  2021-05-12 11:05:30,024 INFO [kafka.zookeeper.ZooKeeperClient][main] [] [ZooKeeperClient Kafka server] Initializing a new session to localhost:2181.
  2021-05-12 11:05:30,027 INFO [org.apache.zookeeper.ZooKeeper][main] [] Initiating client connection, connectString=localhost:2181 sessionTimeout=6000 watcher=kafka.zookeeper.ZooKeeperClient$ZooKeeperClientWatcher$@29bc4bf4
  2021-05-12 11:05:30,027 INFO [org.apache.zookeeper.ClientCnxnSocket][main] [] jute.maxbuffer value is 4194304 Bytes
  2021-05-12 11:05:30,028 INFO [org.apache.zookeeper.ClientCnxn][main] [] zookeeper.request.timeout value is 0. feature enabled=
  2021-05-12 11:05:30,034 INFO [kafka.zookeeper.ZooKeeperClient][main] [] [ZooKeeperClient Kafka server] Waiting until connected.
  2021-05-12 11:05:30,035 INFO [org.apache.zookeeper.ClientCnxn][main-SendThread(localhost:2181)] [] Opening socket connection to server localhost/127.0.0.1:2181
  2021-05-12 11:05:30,036 INFO [org.apache.zookeeper.ClientCnxn][main-SendThread(localhost:2181)] [] Socket connection established, initiating session, client: /127.0.0.1:59560, server: localhost/127.0.0.1:2181
  2021-05-12 11:05:30,037 ERROR [org.apache.zookeeper.server.ZooKeeperSaslServer][NIOServerCxnFactory.SelectorThread-0] [] server principal name/hostname determination error: 
  java.lang.StringIndexOutOfBoundsException: String index out of range: -1
      at java.lang.String.substring(String.java:1967)
      at org.apache.zookeeper.util.SecurityUtils.createSaslServer(SecurityUtils.java:174)
      at org.apache.zookeeper.server.ZooKeeperSaslServer.createSaslServer(ZooKeeperSaslServer.java:44)
      at org.apache.zookeeper.server.ZooKeeperSaslServer.&lt;init&gt;(ZooKeeperSaslServer.java:38)
      at org.apache.zookeeper.server.NIOServerCnxn.&lt;init&gt;(NIOServerCnxn.java:104)
      at org.apache.zookeeper.server.NIOServerCnxnFactory.createConnection(NIOServerCnxnFactory.java:848)
      at org.apache.zookeeper.server.NIOServerCnxnFactory$SelectorThread.processAcceptedConnections(NIOServerCnxnFactory.java:479)
      at org.apache.zookeeper.server.NIOServerCnxnFactory$SelectorThread.run(NIOServerCnxnFactory.java:392)
  2021-05-12 11:05:30,056 INFO [org.apache.zookeeper.ClientCnxn][main-SendThread(localhost:2181)] [] Session establishment complete on server localhost/127.0.0.1:2181, sessionid = 0x10143dd4de40008, negotiated timeout = 6000
  2021-05-12 11:05:30,065 INFO [kafka.zookeeper.ZooKeeperClient][main] [] [ZooKeeperClient Kafka server] Connected.
  2021-05-12 11:05:30,478 ERROR [kafka.server.KafkaServer][main] [] Fatal error during KafkaServer startup. Prepare to shutdown
  java.lang.NullPointerException
      at com.fasterxml.jackson.databind.type.ResolvedRecursiveType.equals(ResolvedRecursiveType.java:103)
      at com.fasterxml.jackson.databind.type.TypeBindings$AsKey.equals(TypeBindings.java:458)
      at java.util.concurrent.ConcurrentHashMap.get(ConcurrentHashMap.java:947)
      at com.fasterxml.jackson.databind.util.LRUMap.get(LRUMap.java:68)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromClass(TypeFactory.java:1211)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromParamType(TypeFactory.java:1384)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromAny(TypeFactory.java:1154)
      at com.fasterxml.jackson.databind.type.TypeFactory._resolveSuperInterfaces(TypeFactory.java:1298)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromClass(TypeFactory.java:1243)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromParamType(TypeFactory.java:1384)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromAny(TypeFactory.java:1154)
      at com.fasterxml.jackson.databind.type.TypeFactory._resolveSuperInterfaces(TypeFactory.java:1298)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromClass(TypeFactory.java:1243)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromParamType(TypeFactory.java:1384)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromAny(TypeFactory.java:1154)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromParamType(TypeFactory.java:1380)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromAny(TypeFactory.java:1154)
      at com.fasterxml.jackson.databind.type.TypeFactory._resolveSuperInterfaces(TypeFactory.java:1298)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromClass(TypeFactory.java:1243)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromParamType(TypeFactory.java:1384)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromAny(TypeFactory.java:1154)
      at com.fasterxml.jackson.databind.type.TypeFactory._resolveSuperInterfaces(TypeFactory.java:1298)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromClass(TypeFactory.java:1243)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromParamType(TypeFactory.java:1384)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromAny(TypeFactory.java:1154)
      at com.fasterxml.jackson.databind.type.TypeFactory._resolveSuperInterfaces(TypeFactory.java:1298)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromClass(TypeFactory.java:1243)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromParamType(TypeFactory.java:1384)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromAny(TypeFactory.java:1154)
      at com.fasterxml.jackson.databind.type.TypeFactory._resolveSuperInterfaces(TypeFactory.java:1298)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromClass(TypeFactory.java:1243)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromParamType(TypeFactory.java:1384)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromAny(TypeFactory.java:1154)
      at com.fasterxml.jackson.databind.type.TypeFactory._resolveSuperInterfaces(TypeFactory.java:1298)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromClass(TypeFactory.java:1243)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromParamType(TypeFactory.java:1384)
      at com.fasterxml.jackson.databind.type.TypeFactory._fromAny(TypeFactory.java:1154)
      at com.fasterxml.jackson.databind.type.TypeFactory.constructType(TypeFactory.java:622)
      at com.fasterxml.jackson.databind.introspect.AnnotatedClass.resolveType(AnnotatedClass.java:228)
      at com.fasterxml.jackson.databind.introspect.AnnotatedConstructor.getParameterType(AnnotatedConstructor.java:104)
      at com.fasterxml.jackson.databind.introspect.AnnotatedWithParams.getParameter(AnnotatedWithParams.java:105)
      at com.fasterxml.jackson.databind.introspect.POJOPropertiesCollector._addCreators(POJOPropertiesCollector.java:449)
      at com.fasterxml.jackson.databind.introspect.POJOPropertiesCollector.collectAll(POJOPropertiesCollector.java:303)
      at com.fasterxml.jackson.databind.introspect.POJOPropertiesCollector.getJsonValueMethod(POJOPropertiesCollector.java:172)
      at com.fasterxml.jackson.databind.introspect.BasicBeanDescription.findJsonValueMethod(BasicBeanDescription.java:224)
      at com.fasterxml.jackson.databind.ser.BasicSerializerFactory.findSerializerByAnnotations(BasicSerializerFactory.java:350)
      at com.fasterxml.jackson.databind.ser.BasicSerializerFactory.buildMapSerializer(BasicSerializerFactory.java:763)
      at com.fasterxml.jackson.databind.ser.BasicSerializerFactory.buildContainerSerializer(BasicSerializerFactory.java:572)
      at com.fasterxml.jackson.databind.ser.BeanSerializerFactory._createSerializer2(BeanSerializerFactory.java:190)
      at com.fasterxml.jackson.databind.ser.BeanSerializerFactory.createSerializer(BeanSerializerFactory.java:159)
      at com.fasterxml.jackson.databind.SerializerProvider._createUntypedSerializer(SerializerProvider.java:1272)
      at com.fasterxml.jackson.databind.SerializerProvider._createAndCacheUntypedSerializer(SerializerProvider.java:1222)
      at com.fasterxml.jackson.databind.SerializerProvider.findValueSerializer(SerializerProvider.java:499)
      at com.fasterxml.jackson.databind.SerializerProvider.findTypedValueSerializer(SerializerProvider.java:697)
      at com.fasterxml.jackson.databind.ser.DefaultSerializerProvider.serializeValue(DefaultSerializerProvider.java:270)
      at com.fasterxml.jackson.databind.ObjectMapper._configAndWriteValue(ObjectMapper.java:3672)
      at com.fasterxml.jackson.databind.ObjectMapper.writeValueAsBytes(ObjectMapper.java:3072)
      at kafka.utils.Json$.encodeAsBytes(Json.scala:115)
      at kafka.zk.ClusterIdZNode$.toJson(ZkData.scala:709)
      at kafka.zk.KafkaZkClient.createOrGetClusterId(KafkaZkClient.scala:1524)
      at kafka.server.KafkaServer$$anonfun$getOrGenerateClusterId$1.apply(KafkaServer.scala:404)
      at kafka.server.KafkaServer$$anonfun$getOrGenerateClusterId$1.apply(KafkaServer.scala:404)
      at scala.Option.getOrElse(Option.scala:121)
      at kafka.server.KafkaServer.getOrGenerateClusterId(KafkaServer.scala:404)
      at kafka.server.KafkaServer.startup(KafkaServer.scala:210)
      at kafka.server.KafkaServerStartable.startup(KafkaServerStartable.scala:44)
  </code></pre>
  <p>Tried debugging the class, found that the code throws the exception when it is trying to get the cluster id of the already <strong>connected</strong> (as per the logs) Zookeeper.</p>