private IConfig createBrokerConfig(final PipeParameters pipeParameters) {
    final Properties properties = new Properties();
    properties.setProperty(
        BrokerConstants.HOST_PROPERTY_NAME,
        pipeParameters.getStringOrDefault(
            PipeExtractorConstant.MQTT_BROKER_HOST_KEY,
            PipeExtractorConstant.MQTT_BROKER_HOST_DEFAULT_VALUE));
    properties.setProperty(
        BrokerConstants.PORT_PROPERTY_NAME,
        pipeParameters.getStringOrDefault(
            PipeExtractorConstant.MQTT_BROKER_PORT_KEY,
            PipeExtractorConstant.MQTT_BROKER_PORT_DEFAULT_VALUE));
    properties.setProperty(
        BrokerConstants.BROKER_INTERCEPTOR_THREAD_POOL_SIZE,
        pipeParameters.getStringOrDefault(
            PipeExtractorConstant.MQTT_BROKER_INTERCEPTOR_THREAD_POOL_SIZE_KEY,
            String.valueOf(
                PipeExtractorConstant.MQTT_BROKER_INTERCEPTOR_THREAD_POOL_SIZE_DEFAULT_VALUE)));
    properties.setProperty(
        BrokerConstants.DATA_PATH_PROPERTY_NAME,
        pipeParameters.getStringOrDefault(
            PipeExtractorConstant.MQTT_DATA_PATH_PROPERTY_NAME_KEY,
            PipeExtractorConstant.MQTT_DATA_PATH_PROPERTY_NAME_DEFAULT_VALUE));
    properties.setProperty(
        BrokerConstants.IMMEDIATE_BUFFER_FLUSH_PROPERTY_NAME,
        pipeParameters.getStringOrDefault(
            PipeExtractorConstant.MQTT_IMMEDIATE_BUFFER_FLUSH_PROPERTY_NAME_KEY,
            String.valueOf(
                PipeExtractorConstant.MQTT_IMMEDIATE_BUFFER_FLUSH_PROPERTY_NAME_DEFAULT_VALUE)));
    properties.setProperty(
        BrokerConstants.ALLOW_ANONYMOUS_PROPERTY_NAME,
        pipeParameters.getStringOrDefault(
            PipeExtractorConstant.MQTT_ALLOW_ANONYMOUS_PROPERTY_NAME_KEY,
            String.valueOf(
                PipeExtractorConstant.MQTT_ALLOW_ANONYMOUS_PROPERTY_NAME_DEFAULT_VALUE)));
    properties.setProperty(
        BrokerConstants.ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME,
        pipeParameters.getStringOrDefault(
            PipeExtractorConstant.MQTT_ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME_KEY,
            String.valueOf(
                PipeExtractorConstant.MQTT_ALLOW_ZERO_BYTE_CLIENT_ID_PROPERTY_NAME_DEFAULT_VALUE)));
    properties.setProperty(
        BrokerConstants.NETTY_MAX_BYTES_PROPERTY_NAME,
        pipeParameters.getStringOrDefault(
            PipeExtractorConstant.MQTT_NETTY_MAX_BYTES_PROPERTY_NAME_KEY,
            String.valueOf(
                PipeExtractorConstant.MQTT_NETTY_MAX_BYTES_PROPERTY_NAME_DEFAULT_VALUE)));
    return new MemoryConfig(properties);
  }
MQTTExtractor is an external Extractor that uses the MQTT protocol to receive data. It starts an
MQTT broker and listens for incoming messages, which are then processed and passed to the pending
queue.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/mqtt/MQTTExtractor.java