We currently have a pipeline with 2 Sinks.
- S3 Sink
- Elastic Sink
We are consuming messages from Kafka and after modifying a field to the Events Based on Some Rules, messages are published to both S3 and Elastic.
We have come Across a Usecase where we had to send Events to two different Index Within the Same Elastic Cluster.
While trying this We Observed that the messages are being sent to only one of the Index.
ElasticSink :
ElasticsearchSink.Builder<Output> esSinkBuilder = new ElasticsearchSink.Builder<Output>(
elasticSearchHosts,
new ElasticsearchSinkFunction<Output>() {
public IndexRequest createIndexRequest(Output output) {
IndexRequest request = Requests.indexRequest()
.index(...)
.source(GsonUtils.toJsonFromObject(output), XContentType.JSON);
return request;
}
@Override
public void process(Output output, RuntimeContext ctx, RequestIndexer indexer) {
indexer.add(createIndexRequest(output));
}
}
);
esSinkBuilder.setRestClientFactory(restClientBuilder -> {
restClientBuilder.setHttpClientConfigCallback(httpClientBuilder -> {
TrustStrategy acceptingTrustStrategy = (certificate, authType) -> true;
SSLContext sslContext = null;
try {
sslContext = SSLContexts.custom().loadTrustMaterial(null, acceptingTrustStrategy).build();
} catch (Exception e) {
// Handle error
}
httpClientBuilder.setSSLContext(sslContext);
CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials("any","any"));
return httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
});
});
esSinkBuilder.setFailureHandler(new ActionRequestFailureHandler() {
@Override
public void onFailure(ActionRequest action, Throwable failure, int restStatusCode, RequestIndexer indexer) {
....
}
});
return esSinkBuilder.build();
}I am unable to see a reason why this is happening. While Debug I can see both sink transformations being added to env.
i am using flink 1.13.x .