code: public void updateKryoSerializer() {
          WorkerTopologyContext workerTopologyContext = contextMaker.makeWorkerTopologyContext(sysTopology);
          KryoTupleDeserializer kryoTupleDeserializer = new KryoTupleDeserializer(stormConf, workerTopologyContext, workerTopologyContext.getRawTopology());
          KryoTupleSerializer kryoTupleSerializer = new KryoTupleSerializer(stormConf, workerTopologyContext.getRawTopology());
  
          atomKryoDeserializer.getAndSet(kryoTupleDeserializer);
          atomKryoSerializer.getAndSet(kryoTupleSerializer);
      }
docs: create kryo serializer