in_cloudwatch log messages getting dropped

[sasi1124]

Problem

I'm using cloudwatch_in to fetch logs from cloudwatch and publishing them to kafka topic. But I see that some of the log messages are missing in kafka topic. I cannot tell whether the messages are getting dropped while pushing it to kafka topic or the plugin is not scrapping them. I see "Rate exceeded" errors in the logs while connecting to cloudwatch and I'm working with Amazon to increase the limits on the cloudwatch for our account.
Can someone tell me if the plugin will retry when it gets "Rate exceeded" error and is this the reason for missing log messages?

...

Steps to replicate

     <source>
      @type cloudwatch_logs
      aws_key_id {{ aws_access_key_id}}
      aws_sec_key {{ aws_secret_access_key}}
      region {{ aws_region}}
      tag ordercapture-cloudwatch.out
      log_group_name /aws/lambda/ordercapture-preprod_api
      use_log_stream_name_prefix false
      use_todays_log_stream true
      fetch_interval 30
      state_file /var/tmp/ordercapture-preprod_api.out.state
      json_handler json
      format none
    </source>
    <source>
      @type cloudwatch_logs
      aws_key_id {{ aws_access_key_id}}
      aws_sec_key {{ aws_secret_access_key}}
      region {{ aws_region}}
      tag ordercapture-cloudwatch.out
      log_group_name /aws/lambda/ordercapture-preprod_authorizer
      use_log_stream_name_prefix false
      use_todays_log_stream true
      fetch_interval 30
      state_file /var/tmp/ordercapture-preprod_authorizer.out.state
      json_handler json
      format none
    </source>
    <source>
      @type cloudwatch_logs
      aws_key_id {{ aws_access_key_id}}
      aws_sec_key {{ aws_secret_access_key}}
      region {{ aws_region}}
      tag ordercapture-cloudwatch.out
      log_group_name /aws/lambda/ordercapture-preprod_csv_convert
      use_log_stream_name_prefix false
      use_todays_log_stream true
      fetch_interval 30
      state_file /var/tmp/ordercapture-preprod_csv_convert.out.state
      json_handler json
      format none
    </source>
    <source>
      @type cloudwatch_logs
      aws_key_id {{ aws_access_key_id}}
      aws_sec_key {{ aws_secret_access_key}}
      region {{ aws_region}}
      tag ordercapture-cloudwatch.out
      log_group_name /aws/lambda/ordercapture-preprod_execution
      use_log_stream_name_prefix false
      use_todays_log_stream true
      fetch_interval 30
      state_file /var/tmp/ordercapture-preprod_execution.out.state
      json_handler json
      format none
    </source>
    <filter *.**>
      @type record_transformer
      <record>
      topic ${tag_prefix[0]}
      </record>
    </filter>

    <match *.**>
       @type               kafka_buffered
       brokers             {{ broker }}
       default_topic       topic
       output_data_type    json
       ssl_ca_cert         {{ sslcert}}

       <buffer>
             @type memory
             #flush_interval 1s
             flush_at_shutdown true
             retry_max_times 5
             flush_mode immediate
             #path /var/log/fluent/s3/ingest
             #timekey 3600 # 1 hour partition
             #timekey_wait 10m
             #timekey_use_utc true # use utc
             #chunk_limit_size 128m
       </buffer>
    </match>

Expected Behavior or What you need to ask

All the log messages from above mentioned streams should be pushed to kafka topic

Using Fluentd and CloudWatchLogs plugin versions

• OS version

NAME="Alpine Linux"
ID=alpine
VERSION_ID=3.8.1
PRETTY_NAME="Alpine Linux v3.8"

• Bare Metal or within Docker or Kubernetes or others?
  Kubernetes

• Fluentd v1.3.2

• Fluent-gem list

activesupport (5.2.3)
addressable (2.6.0)
aws-eventstream (1.0.3)
aws-partitions (1.156.0)
aws-sdk-cloudwatchlogs (1.17.0)
aws-sdk-core (3.49.0)
aws-sigv4 (1.1.0)
bigdecimal (1.3.5)
cmath (default: 1.0.0)
concurrent-ruby (1.1.5)
cool.io (1.5.3)
csv (default: 1.0.0)
date (default: 1.0.0)
dig_rb (1.0.1)
digest-crc (0.4.1)
domain_name (0.5.20180417)
etc (default: 1.0.0)
fcntl (default: 1.0.0)
fileutils (default: 1.0.2)
fluent-plugin-cloudwatch-logs (0.7.3)
fluent-plugin-kafka (0.9.2)
fluent-plugin-kubernetes_metadata_filter (2.1.6)
fluentd (1.3.2)
http (0.9.8)
http-cookie (1.0.3)
http-form_data (1.0.3)
http_parser.rb (0.6.0)
i18n (1.6.0)
ipaddr (default: 1.2.0)
jmespath (1.4.0)
json (2.1.0)
kubeclient (1.1.4)
lru_redux (1.1.0)
ltsv (0.1.2)
mime-types (3.2.2)
mime-types-data (3.2019.0331)
minitest (5.11.3)
msgpack (1.2.4)
netrc (0.11.0)
oj (3.3.10)
openssl (default: 2.1.2)
psych (default: 3.0.2)
public_suffix (3.0.3)
recursive-open-struct (1.0.0)
rest-client (2.0.2)
ruby-kafka (0.7.6)
scanf (default: 1.0.0)
serverengine (2.1.0)
sigdump (0.2.4)
stringio (default: 0.0.1)
strptime (0.2.3)
strscan (default: 1.0.0)
thread_safe (0.3.6)
tzinfo (1.2.5)
tzinfo-data (1.2018.7)
unf (0.1.4)
unf_ext (0.0.7.6)
webrick (default: 1.4.2)
yajl-ruby (1.4.1)
zlib (default: 1.0.0)