code: @Override
      public void connect() {
          if (connected) {
              return;
          }
  
          connected = true;
          if (kafkaConsumer == null && !flatMessage) {
              kafkaConsumer = new KafkaConsumer<String, Message>(properties);
  
          }
          if (kafkaConsumer2 == null && flatMessage) {
              kafkaConsumer2 = new KafkaConsumer<String, String>(properties);
          }
      }
docs: 打开连接