flink 1.12 的elasticsearch sink 如何设置用户名密码

在 Flink 1.12 的 Elasticsearch sink 中，可以通过配置 Elasticsearch 的用户名和密码来连接 Elasticsearch 集群。具体的配置方式如下：

1. 在 Flink 程序中引入 Elasticsearch 依赖，例如：

<dependency>
  <groupId>org.apache.flink</groupId>
  <artifactId>flink-connector-elasticsearch7_2.12</artifactId>
  <version>1.12.0</version>
</dependency>

2. 在 Flink 程序中创建 Elasticsearch 的连接信息，例如：

Map<String, String> config = new HashMap<>();
config.put("cluster.name", "myClusterName");
config.put("bulk.flush.max.actions", "1");
config.put("index", "myIndexName");
config.put("hostname", "localhost");
config.put("port", "9200");
config.put("security.enable.ssl", "false");
config.put("security.username", "myUserName");
config.put("security.password", "myPassword");

其中，`security.username` 和 `security.password` 分别表示 Elasticsearch 的用户名和密码。

3. 在 Flink 程序中创建 Elasticsearch 的 Sink，例如：

DataStream<String> stream = ...;
stream.addSink(new ElasticsearchSink<>(config, new MyElasticsearchSinkFunction()));

其中，`MyElasticsearchSinkFunction` 表示自定义的 Elasticsearch Sink Function。

通过以上步骤，即可在 Flink 1.12 的 Elasticsearch sink 中设置用户名和密码。