[Bug]: ElasticsearchIO should use RetryPolicy in case of connectivity issues

[bvolpato]

What happened?

ElasticsearchIO has proper handling for 429 Too Many Requests, but high load to Elasticsearch may be expressed through several other symptoms.

For example, I've noticed Connection reset and Broken pipe. Failing the bundle may impact the pipeline negatively, when the intent of the user is just to try the Elasticsearch operation.

It would be great to revisit the default RetryPredicate to also retry in those connectivity issues.

Issue Priority

Priority: 3 (minor)

Issue Components

• Component: Python SDK
• Component: Java SDK
• Component: Go SDK
• Component: Typescript SDK
• Component: IO connector
• Component: Beam YAML
• Component: Beam examples
• Component: Beam playground
• Component: Beam katas
• Component: Website
• Component: Spark Runner
• Component: Flink Runner
• Component: Samza Runner
• Component: Twister2 Runner
• Component: Hazelcast Jet Runner
• Component: Google Cloud Dataflow Runner

[bvolpato]

Stacktrace for one of the issues

java.io.IOException: Connection reset by peer
	at org.elasticsearch.client.RestClient.extractAndWrapCause(RestClient.java:935)
	at org.elasticsearch.client.RestClient.performRequest(RestClient.java:300)
	at org.elasticsearch.client.RestClient.performRequest(RestClient.java:288)
	at org.apache.beam.sdk.io.elasticsearch.ElasticsearchIO$BulkIO$BulkIOBaseFn.flushBatch(ElasticsearchIO.java:2550)
	at org.apache.beam.sdk.io.elasticsearch.ElasticsearchIO$BulkIO$BulkIOBaseFn.flushAndOutputResults(ElasticsearchIO.java:2483)
	at org.apache.beam.sdk.io.elasticsearch.ElasticsearchIO$BulkIO$BulkIOBaseFn.finishBundle(ElasticsearchIO.java:2444)
Caused by: java.io.IOException: Connection reset by peer
	at java.base/sun.nio.ch.FileDispatcherImpl.read0(Native Method)
	at java.base/sun.nio.ch.SocketDispatcher.read(SocketDispatcher.java:39)
	at java.base/sun.nio.ch.IOUtil.readIntoNativeBuffer(IOUtil.java:276)
	at java.base/sun.nio.ch.IOUtil.read(IOUtil.java:245)
	at java.base/sun.nio.ch.IOUtil.read(IOUtil.java:223)
	at java.base/sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:358)
	at org.apache.http.impl.nio.reactor.SessionInputBufferImpl.fill(SessionInputBufferImpl.java:231)