target.loggerOnce(context.Background(), err, target.ID().String())
		}
		return err
	}

	if target.natsConn != nil && args.JetStream.Enable {
		var jstream nats.JetStream
		jstream, err = target.natsConn.JetStream()
		if err != nil {
			if err.Error() != nats.ErrNoServers.Error() {
				target.loggerOnce(context.Background(), err, target.ID().String())
			}
			return err
		}

			Description: "client cert key for NATS mTLS auth",
			Optional:    true,
			Type:        "string",
			Sensitive:   true,
		},
		config.HelpKV{
			Key:         target.NATSJetStream,
			Description: "enable JetStream support",
			Optional:    true,
			Type:        "on|off",
		},
		config.HelpKV{
			Key:         target.NATSQueueDir,
			Description: queueDirComment,
			Optional:    true,
			Type:        "path",

			PingInterval:    pingInterval,
			QueueDir:        env.Get(queueDirEnv, kv.Get(target.NATSQueueDir)),
			QueueLimit:      queueLimit,
			RootCAs:         rootCAs,
		}
		natsArgs.JetStream.Enable = env.Get(jetStreamEnableEnv, kv.Get(target.NATSJetStream)) == config.EnableOn

		streamingEnableEnv := target.EnvNATSStreaming
		if k != config.Default {

* Add external link [Apache Kafka producer and consumer with FastAPI and aiokafka](https://iwpnd.pw/articles/2020-03/apache-kafka-fastapi-geostream) to docs. PR [#1112](https://github.com/tiangolo/fastapi/pull/1112) by [@iwpnd](https://github.com/iwpnd).