429 Too Many Requests /_bulk

#### Issue
Initially, Fluentd is working file for some days,
but after some time facing this issue and Fluentd doesn't send logs to OpenSearch
when I switched from [fluent-plugin-opensearch](https://github.com/fluent/fluent-plugin-opensearch) to [fluent-plugin-aws-elasticsearch-service](https://github.com/atomita/fluent-plugin-aws-elasticsearch-service) and buffer type as memory I didn't faced this issue.
```

@type "file"
path "/var/log/fluentd-buffers/kubernetes-jb-event-txn.system.buffer"
flush_mode interval
flush_interval 1s
flush_thread_count 4
chunk_full_threshold 0.9

```
```

@type memory

```

#### Issue log
```
2022-02-21 15:43:50 +0000 [debug]: #0 taking back chunk for errors. chunk="5d88918bc47758f16ab54ea9a55a8407"
2022-02-21 15:43:50 +0000 [warn]: #0 failed to flush the buffer. retry_times=0 next_retry_time=2022-02-21 15:43:53 +0000 chunk="5d88918bc47758f16ab54ea9a55a8407" error_class=Fluent::Plugin::OpenSearchOutput::RecoverableRequestFailure error="could not push logs to OpenSearch cluster ({:host=>\"vpc-east-1.es.amazonaws.com\", :port=>443, :scheme=>\"https\", :user=>\"admin\", :password=>\"obfuscated\"}): [429] 429 Too Many Requests /_bulk"
2022-02-21 15:43:50 +0000 [warn]: #0 /fluentd/vendor/bundle/ruby/2.6.0/gems/fluent-plugin-opensearch-1.0.1/lib/fluent/plugin/out_opensearch.rb:1101:in `rescue in send_bulk'
2022-02-21 15:43:50 +0000 [warn]: #0 /fluentd/vendor/bundle/ruby/2.6.0/gems/fluent-plugin-opensearch-1.0.1/lib/fluent/plugin/out_opensearch.rb:1063:in `send_bulk'
2022-02-21 15:43:50 +0000 [warn]: #0 /fluentd/vendor/bundle/ruby/2.6.0/gems/fluent-plugin-opensearch-1.0.1/lib/fluent/plugin/out_opensearch.rb:878:in `block in write'
2022-02-21 15:43:50 +0000 [warn]: #0 /fluentd/vendor/bundle/ruby/2.6.0/gems/fluent-plugin-opensearch-1.0.1/lib/fluent/plugin/out_opensearch.rb:877:in `each'
2022-02-21 15:43:50 +0000 [warn]: #0 /fluentd/vendor/bundle/ruby/2.6.0/gems/fluent-plugin-opensearch-1.0.1/lib/fluent/plugin/out_opensearch.rb:877:in `write'
2022-02-21 15:43:50 +0000 [warn]: #0 /fluentd/vendor/bundle/ruby/2.6.0/gems/fluentd-1.14.3/lib/fluent/plugin/output.rb:1179:in `try_flush'
2022-02-21 15:43:50 +0000 [warn]: #0 /fluentd/vendor/bundle/ruby/2.6.0/gems/fluentd-1.14.3/lib/fluent/plugin/output.rb:1491:in `flush_thread_run'
2022-02-21 15:43:50 +0000 [warn]: #0 /fluentd/vendor/bundle/ruby/2.6.0/gems/fluentd-1.14.3/lib/fluent/plugin/output.rb:499:in `block (2 levels) in start'
2022-02-21 15:43:50 +0000 [warn]: #0 /fluentd/vendor/bundle/ruby/2.6.0/gems/fluentd-1.14.3/lib/fluent/plugin_helper/thread.rb:78:in `block in thread_create'
2022-02-21 15:43:52 +0000 [warn]: #0 retry succeeded. chunk_id="5d88918bc47758f16ab54ea9a55a8407"
2022-02-21 15:43:58 +0000 [debug]: #0 Created new chunk chunk_id="5d88919cf1b1dbf266ce66e8c93f6df4" metadata=#
2022-02-21 15:44:16 +0000 [debug]: #0 Created new chunk chunk_id="5d8891ae20521b741c6002d1f72f4f2d" metadata=#
```

#### Steps to replicate
Config file used to connect to AWS OpenSearch
```
2022-02-23 02:32:29 +0000 [info]: using configuration file:

@type tail
read_from_head true
tag "kubernetes"
path "/var/log/containers/*java*container*.log"
path_key "path"
pos_file "/var/log/fluentd-containers.log.pos"
exclude_path ["/var/log/containers/fluent*"]

@type "json"
json_parser json
time_key "time"
time_format "%iso8601"
unmatched_lines
time_type string



@type kafka_group
brokers "b-1.amazonaws.com:9092,b-2.amazonaws.com:9092"
consumer_group "amazon.broker-2"
topics "jb-audit-user"
format "json"


@type kafka_group
brokers "b-1.amazonaws.com:9092,b-2.amazonaws.com:9092"
consumer_group "amazon.broker-2"
topics "jb-audit-sys"
format "json"


@type kafka_group
brokers "b-1.amazonaws.com:9092,b-2.amazonaws.com:9092"
consumer_group "amazon.broker-2"
topics "MT-Incoming-payments"
format "json"


@type parser
key_name "log"
reserve_time true
reserve_data true
remove_key_name_field true
replace_invalid_sequence true
emit_invalid_record_to_error true

@type "multi_format"

format json


format none




@type opensearch
ssl_verify false
@log_level "debug"
logstash_format true
logstash_prefix "jb-app-log-java"
logstash_prefix_separator "-"
logstash_dateformat "%Y.%m"
user "admin"
password xxxxxx

url https://vpc.amazonaws.com:443
region "us-east-1"


@type "file"
path "/var/log/fluentd-buffers/kubernetes.system.buffer"
flush_mode interval
flush_interval 10s
flush_thread_count 8
flush_at_shutdown true
chunk_full_threshold 0.9
retry_forever true
retry_type exponential_backoff
retry_wait 2s



@type opensearch
ssl_verify false
@log_level "debug"
logstash_format true
logstash_prefix "jb-audit-user"
logstash_prefix_separator "-"
logstash_dateformat "%Y.%m"
user "admin"
password xxxxxx

url https://vpc.amazonaws.com:443
region "us-east-1"


@type "file"
path "/var/log/fluentd-buffers/kubernetes-jb-audit-user.system.buffer"
flush_mode interval
flush_interval 10s
flush_thread_count 8
flush_at_shutdown true
chunk_full_threshold 0.9
retry_forever true
retry_type exponential_backoff
retry_wait 2s



@type opensearch
ssl_verify false
@log_level "debug"
logstash_format true
logstash_prefix "jb-audit-sys"
logstash_prefix_separator "-"
logstash_dateformat "%Y.%m"
user "admin"
password xxxxxx

url https://vpc.amazonaws.com:443
region "us-east-1"


@type "file"
path "/var/log/fluentd-buffers/kubernetes-jb-audit-sys.system.buffer"
flush_mode interval
flush_interval 10s
flush_thread_count 8
flush_at_shutdown true
chunk_full_threshold 0.9
retry_forever true
retry_type exponential_backoff
retry_wait 2s



@type opensearch
ssl_verify false
@log_level "debug"
logstash_format true
logstash_prefix "jb-event-txn"
logstash_prefix_separator "-"
logstash_dateformat "%Y.%m"
user "admin"
password xxxxxx

url https://vpc.amazonaws.com:443
region "us-east-1"


@type "memory"

```

#### Expected Behavior or What you need to ask

how to overcome this issue and keep retrying forever if this issue is from OpenSearch
...

#### Using Fluentd and OpenSearch plugin versions

Using Fluentd and OpenSearch plugin versions
2022-02-23 02:32:28 +0000 [info]: gem 'fluent-plugin-avro' version '1.1.1'
2022-02-23 02:32:28 +0000 [info]: gem 'fluent-plugin-concat' version '2.5.0'
2022-02-23 02:32:28 +0000 [info]: gem 'fluent-plugin-detect-exceptions' version '0.0.14'
2022-02-23 02:32:28 +0000 [info]: gem 'fluent-plugin-grok-parser' version '2.6.2'
2022-02-23 02:32:28 +0000 [info]: gem 'fluent-plugin-json-in-json-2' version '1.0.2'
2022-02-23 02:32:28 +0000 [info]: gem 'fluent-plugin-kafka' version '0.17.3'
2022-02-23 02:32:28 +0000 [info]: gem 'fluent-plugin-kubernetes_metadata_filter' version '2.9.2'
2022-02-23 02:32:28 +0000 [info]: gem 'fluent-plugin-multi-format-parser' version '1.0.0'
2022-02-23 02:32:28 +0000 [info]: gem 'fluent-plugin-opensearch' version '1.0.1'
2022-02-23 02:32:28 +0000 [info]: gem 'fluent-plugin-parser-avro' version '0.3.1'
2022-02-23 02:32:28 +0000 [info]: gem 'fluent-plugin-parser-cri' version '0.1.1'
2022-02-23 02:32:28 +0000 [info]: gem 'fluent-plugin-prometheus' version '2.0.2'
2022-02-23 02:32:28 +0000 [info]: gem 'fluent-plugin-record-modifier' version '2.1.0'
2022-02-23 02:32:28 +0000 [info]: gem 'fluent-plugin-rewrite-tag-filter' version '2.4.0'
2022-02-23 02:32:28 +0000 [info]: gem 'fluent-plugin-stdout-pp' version '0.2.0'
2022-02-23 02:32:28 +0000 [info]: gem 'fluent-plugin-systemd' version '1.0.5'
2022-02-23 02:32:28 +0000 [info]: gem 'fluentd' version '1.14.3'
2022-02-21 13:50:38 +0000 [info]: starting fluentd-1.14.3 pid=7 ruby="2.6.9"

Start with lib/fluent/plugin/out_opensearch.rb at send_bulk (lines 1063 and 1101), then compare the logged 429 response with the configured file-buffer retry settings. Reproduce using the supplied Fluentd 1.14.3 and fluent-plugin-opensearch 1.0.1 configuration, and verify that a 429 remains buffered and is retried without data loss.

Written by the indexing model from the issue text.