diff --git a/CHANGELOG.md b/CHANGELOG.md
index 56f8ba42ec..3baf9cfe86 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,6 +6,10 @@
 
   The agent now supports Ruby 3.4.0. We've made incremental changes throughout the preview stage to reach compatibility. This release includes an update to the Thread Profiler for compatibility with Ruby 3.4.0's new backtrace format. [Issue#2992](https://github.com/newrelic/newrelic-ruby-agent/issues/2992) [PR#2997](https://github.com/newrelic/newrelic-ruby-agent/pull/2997)
 
+- **Feature: Add instrumentation for aws-sdk-kinesis**
+
+  The agent now has instrumentation for the [aws-sdk-kinesis](https://rubygems.org/gems/aws-sdk-kinesis) gem. It will record message broker segments for `get_records`, `put_record`, and `put_records` operations. All other operations will record standard segments. [PR#2974](https://github.com/newrelic/newrelic-ruby-agent/pull/2974)
+
 - **Bugfix: Stop emitting inaccurate debug-level log about deprecated configuration options**
 
   In the previous major release, we dropped support for `disable_<library_name>` configuration options in favor of `instrumentation.<library_name>`. Previously, a DEBUG level log warning appeared whenever `disable_*` options were set to `true`, even for libraries (e.g. Action Dispatch) without equivalent `instrumentation.*` options:
diff --git a/lib/new_relic/agent/configuration/default_source.rb b/lib/new_relic/agent/configuration/default_source.rb
index 86e84594af..a87eecadf6 100644
--- a/lib/new_relic/agent/configuration/default_source.rb
+++ b/lib/new_relic/agent/configuration/default_source.rb
@@ -1562,6 +1562,15 @@ def self.notify
           :allowed_from_server => false,
           :description => 'Controls auto-instrumentation of the aws_sdk_lambda library at start-up. May be one of `auto`, `prepend`, `chain`, `disabled`.'
         },
+        :'instrumentation.aws_sdk_kinesis' => {
+          :default => 'auto',
+          :documentation_default => 'auto',
+          :public => true,
+          :type => String,
+          :dynamic_name => true,
+          :allowed_from_server => false,
+          :description => 'Controls auto-instrumentation of the aws-sdk-kinesis library at start-up. May be one of `auto`, `prepend`, `chain`, `disabled`.'
+        },
         :'instrumentation.ruby_kafka' => {
           :default => 'auto',
           :public => true,
diff --git a/lib/new_relic/agent/instrumentation/aws_sdk_kinesis.rb b/lib/new_relic/agent/instrumentation/aws_sdk_kinesis.rb
new file mode 100644
index 0000000000..59d82dd053
--- /dev/null
+++ b/lib/new_relic/agent/instrumentation/aws_sdk_kinesis.rb
@@ -0,0 +1,22 @@
+# This file is distributed under New Relic's license terms.
+# See https://github.com/newrelic/newrelic-ruby-agent/blob/main/LICENSE for complete details.
+# frozen_string_literal: true
+
+require_relative 'aws_sdk_kinesis/instrumentation'
+require_relative 'aws_sdk_kinesis/chain'
+require_relative 'aws_sdk_kinesis/prepend'
+
+DependencyDetection.defer do
+  named :aws_sdk_kinesis
+
+  depends_on do
+    defined?(Aws::Kinesis::Client)
+  end
+  executes do
+    if use_prepend?
+      prepend_instrument Aws::Kinesis::Client, NewRelic::Agent::Instrumentation::Kinesis::Prepend
+    else
+      chain_instrument NewRelic::Agent::Instrumentation::Kinesis::Chain
+    end
+  end
+end
diff --git a/lib/new_relic/agent/instrumentation/aws_sdk_kinesis/chain.rb b/lib/new_relic/agent/instrumentation/aws_sdk_kinesis/chain.rb
new file mode 100644
index 0000000000..78527b1c3a
--- /dev/null
+++ b/lib/new_relic/agent/instrumentation/aws_sdk_kinesis/chain.rb
@@ -0,0 +1,21 @@
+# This file is distributed under New Relic's license terms.
+# See https://github.com/newrelic/newrelic-ruby-agent/blob/main/LICENSE for complete details.
+# frozen_string_literal: true
+
+module NewRelic::Agent::Instrumentation
+  module Kinesis::Chain
+    def self.instrument!
+      ::Aws::Kinesis::Client.class_eval do
+        include NewRelic::Agent::Instrumentation::Kinesis
+
+        NewRelic::Agent::Instrumentation::Kinesis::INSTRUMENTED_METHODS.each do |method_name|
+          alias_method("#{method_name}_without_new_relic".to_sym, method_name.to_sym)
+
+          define_method(method_name) do |*args|
+            instrument_method_with_new_relic(method_name, *args) { send("#{method_name}_without_new_relic".to_sym, *args) }
+          end
+        end
+      end
+    end
+  end
+end
diff --git a/lib/new_relic/agent/instrumentation/aws_sdk_kinesis/instrumentation.rb b/lib/new_relic/agent/instrumentation/aws_sdk_kinesis/instrumentation.rb
new file mode 100644
index 0000000000..c4da1d3c04
--- /dev/null
+++ b/lib/new_relic/agent/instrumentation/aws_sdk_kinesis/instrumentation.rb
@@ -0,0 +1,91 @@
+# This file is distributed under New Relic's license terms.
+# See https://github.com/newrelic/newrelic-ruby-agent/blob/main/LICENSE for complete details.
+# frozen_string_literal: true
+
+module NewRelic::Agent::Instrumentation
+  module Kinesis
+    INSTRUMENTED_METHODS = %w[
+      add_tags_to_stream
+      create_stream
+      decrease_stream_retention_period
+      delete_stream
+      describe_limits
+      describe_stream
+      disable_enhanced_monitoring
+      enable_enhanced_monitoring
+      get_records
+      get_shard_iterator
+      increase_stream_retention_period
+      list_streams
+      list_tags_for_stream
+      merge_shards
+      put_record
+      put_records
+      remove_tags_from_stream
+      split_shard
+      update_shard_count
+    ].freeze
+
+    KINESIS = 'Kinesis'
+    AWS_KINESIS_DATA_STREAMS = 'aws_kinesis_data_streams'
+    MESSAGE_BROKER_SEGMENT_METHODS = %w[put_record put_records get_records].freeze
+
+    def instrument_method_with_new_relic(method_name, *args)
+      return yield unless NewRelic::Agent::Tracer.tracing_enabled?
+
+      NewRelic::Agent.record_instrumentation_invocation(KINESIS)
+      params = args[0]
+      arn = get_arn(params) if params
+
+      if MESSAGE_BROKER_SEGMENT_METHODS.include?(method_name)
+        stream_name = get_stream_name(params, arn)
+        segment = NewRelic::Agent::Tracer.start_message_broker_segment(
+          action: method_name == 'get_records' ? :consume : :produce,
+          library: KINESIS,
+          destination_type: :stream,
+          destination_name: stream_name
+        )
+      else
+        segment = NewRelic::Agent::Tracer.start_segment(name: get_segment_name(method_name, params))
+      end
+
+      segment&.add_agent_attribute('cloud.resource_id', arn) if arn
+
+      begin
+        NewRelic::Agent::Tracer.capture_segment_error(segment) { yield }
+      ensure
+        segment&.add_agent_attribute('cloud.platform', AWS_KINESIS_DATA_STREAMS)
+        segment&.finish
+      end
+    end
+
+    def get_segment_name(method_name, params)
+      stream_name = params&.dig(:stream_name)
+      return "#{KINESIS}/#{method_name}/#{stream_name}" if stream_name
+
+      "#{KINESIS}/#{method_name}"
+    rescue => e
+      NewRelic::Agent.logger.warn("Failed to create segment name: #{e}")
+    end
+
+    def get_stream_name(params, arn)
+      params&.dig(:stream_name) || arn.split('/').last || 'unknown'
+    rescue => e
+      NewRelic::Agent.logger.warn("Failed to get stream name: #{e}")
+    end
+
+    def nr_account_id
+      return @nr_account_id if defined?(@nr_account_id)
+
+      @nr_account_id = NewRelic::Agent::Aws.get_account_id(config)
+    end
+
+    def get_arn(params)
+      stream_arn = params&.dig(:stream_arn)
+      return stream_arn if stream_arn
+
+      stream_name = params&.dig(:stream_name)
+      NewRelic::Agent::Aws.create_arn(KINESIS.downcase, "stream/#{stream_name}", config&.region, nr_account_id) if stream_name
+    end
+  end
+end
diff --git a/lib/new_relic/agent/instrumentation/aws_sdk_kinesis/prepend.rb b/lib/new_relic/agent/instrumentation/aws_sdk_kinesis/prepend.rb
new file mode 100644
index 0000000000..79b2e0bf2e
--- /dev/null
+++ b/lib/new_relic/agent/instrumentation/aws_sdk_kinesis/prepend.rb
@@ -0,0 +1,15 @@
+# This file is distributed under New Relic's license terms.
+# See https://github.com/newrelic/newrelic-ruby-agent/blob/main/LICENSE for complete details.
+# frozen_string_literal: true
+
+module NewRelic::Agent::Instrumentation
+  module Kinesis::Prepend
+    include NewRelic::Agent::Instrumentation::Kinesis
+
+    INSTRUMENTED_METHODS.each do |method_name|
+      define_method(method_name) do |*args|
+        instrument_method_with_new_relic(method_name, *args) { super(*args) }
+      end
+    end
+  end
+end
diff --git a/lib/new_relic/agent/transaction/message_broker_segment.rb b/lib/new_relic/agent/transaction/message_broker_segment.rb
index c511fed283..c11e244f52 100644
--- a/lib/new_relic/agent/transaction/message_broker_segment.rb
+++ b/lib/new_relic/agent/transaction/message_broker_segment.rb
@@ -15,6 +15,7 @@ class MessageBrokerSegment < Segment
         PRODUCE = 'Produce'.freeze
         QUEUE = 'Queue'.freeze
         PURGE = 'Purge'.freeze
+        STREAM = 'Stream'.freeze
         TEMP = 'Temp'.freeze
         TOPIC = 'Topic'.freeze
         UNKNOWN = 'Unknown'.freeze
@@ -22,6 +23,7 @@ class MessageBrokerSegment < Segment
         DESTINATION_TYPES = [
           :exchange,
           :queue,
+          :stream,
           :topic,
           :temporary_queue,
           :temporary_topic,
@@ -37,6 +39,7 @@ class MessageBrokerSegment < Segment
         TYPES = {
           exchange: EXCHANGE,
           temporary_queue: QUEUE,
+          stream: STREAM,
           queue: QUEUE,
           temporary_topic: TOPIC,
           topic: TOPIC,
diff --git a/test/multiverse/suites/aws_sdk_kinesis/Envfile b/test/multiverse/suites/aws_sdk_kinesis/Envfile
new file mode 100644
index 0000000000..1c637ab775
--- /dev/null
+++ b/test/multiverse/suites/aws_sdk_kinesis/Envfile
@@ -0,0 +1,9 @@
+# This file is distributed under New Relic's license terms.
+# See https://github.com/newrelic/newrelic-ruby-agent/blob/main/LICENSE for complete details.
+# frozen_string_literal: true
+
+instrumentation_methods :chain, :prepend
+
+gemfile <<~RB
+  gem 'aws-sdk-kinesis'
+RB
diff --git a/test/multiverse/suites/aws_sdk_kinesis/config/newrelic.yml b/test/multiverse/suites/aws_sdk_kinesis/config/newrelic.yml
new file mode 100644
index 0000000000..dfb4f28ac7
--- /dev/null
+++ b/test/multiverse/suites/aws_sdk_kinesis/config/newrelic.yml
@@ -0,0 +1,19 @@
+---
+development:
+  error_collector:
+    enabled: true
+  apdex_t: 0.5
+  monitor_mode: true
+  license_key: bootstrap_newrelic_admin_license_key_000
+  instrumentation:
+    aws_sdk_kinesis: <%= $instrumentation_method %>
+  app_name: test
+  log_level: debug
+  host: 127.0.0.1
+  api_host: 127.0.0.1
+  transaction_trace:
+    record_sql: obfuscated
+    enabled: true
+    stack_trace_threshold: 0.5
+    transaction_threshold: 1.0
+  capture_params: false
diff --git a/test/multiverse/suites/aws_sdk_kinesis/kinesis_instrumentation_test.rb b/test/multiverse/suites/aws_sdk_kinesis/kinesis_instrumentation_test.rb
new file mode 100644
index 0000000000..f019c813c8
--- /dev/null
+++ b/test/multiverse/suites/aws_sdk_kinesis/kinesis_instrumentation_test.rb
@@ -0,0 +1,532 @@
+# This file is distributed under New Relic's license terms.
+# See https://github.com/newrelic/newrelic-ruby-agent/blob/main/LICENSE for complete details.
+# frozen_string_literal: true
+
+require 'aws-sdk-kinesis'
+
+class KinesisInstrumentationTest < Minitest::Test
+  def setup
+    Aws.config.update(stub_responses: true)
+  end
+
+  def teardown
+    harvest_span_events!
+    NewRelic::Agent.instance.stats_engine.clear_stats
+  end
+
+  def create_client
+    Aws::Kinesis::Client.new(region: 'us-east-2')
+  end
+
+  def test_all_attributes_added_to_segment
+    client = create_client
+
+    NewRelic::Agent::Aws.stub(:create_arn, 'test-arn') do
+      in_transaction do |txn|
+        txn.stubs(:sampled?).returns(true)
+        client.create_stream({
+          stream_name: 'deschutes_river',
+          shard_count: 2
+        })
+      end
+
+      spans = harvest_span_events!
+      span = spans[1][0]
+
+      assert_equal 'Kinesis/create_stream/deschutes_river', span[0]['name']
+      assert_equal 'aws_kinesis_data_streams', span[2]['cloud.platform']
+      assert_equal 'test-arn', span[2]['cloud.resource_id']
+    end
+  end
+
+  def test_delete_stream
+    client = create_client
+
+    NewRelic::Agent::Aws.stub(:create_arn, 'test-arn') do
+      in_transaction do |txn|
+        txn.stubs(:sampled?).returns(true)
+        client.create_stream({
+          stream_name: 'deschutes_river',
+          shard_count: 2
+        })
+
+        client.delete_stream({
+          stream_name: 'deschutes_river'
+        })
+      end
+
+      spans = harvest_span_events!
+      span = spans[1][1]
+
+      assert_equal 'Kinesis/delete_stream/deschutes_river', span[0]['name']
+      assert_equal 'aws_kinesis_data_streams', span[2]['cloud.platform']
+      assert_equal 'test-arn', span[2]['cloud.resource_id']
+    end
+  end
+
+  def test_describe_stream
+    client = create_client
+
+    NewRelic::Agent::Aws.stub(:create_arn, 'test-arn') do
+      in_transaction do |txn|
+        txn.stubs(:sampled?).returns(true)
+        client.create_stream({
+          stream_name: 'deschutes_river',
+          shard_count: 2
+        })
+
+        client.describe_stream({
+          stream_name: 'deschutes_river'
+        })
+      end
+
+      spans = harvest_span_events!
+      span = spans[1][1]
+
+      assert_equal 'Kinesis/describe_stream/deschutes_river', span[0]['name']
+      assert_equal 'aws_kinesis_data_streams', span[2]['cloud.platform']
+      assert_equal 'test-arn', span[2]['cloud.resource_id']
+    end
+  end
+
+  def test_list_streams
+    client = create_client
+
+    NewRelic::Agent::Aws.stub(:create_arn, 'test-arn') do
+      in_transaction do |txn|
+        txn.stubs(:sampled?).returns(true)
+        client.create_stream({
+          stream_name: 'deschutes_river',
+          shard_count: 2
+        })
+
+        client.list_streams
+      end
+
+      spans = harvest_span_events!
+      span = spans[1][1]
+
+      assert_equal 'Kinesis/list_streams', span[0]['name']
+      assert_equal 'aws_kinesis_data_streams', span[2]['cloud.platform']
+      refute span[2]['cloud.resource_id']
+    end
+  end
+
+  def test_add_tags_to_stream
+    client = create_client
+
+    NewRelic::Agent::Aws.stub(:create_arn, 'test-arn') do
+      in_transaction do |txn|
+        txn.stubs(:sampled?).returns(true)
+        client.create_stream({
+          stream_name: 'deschutes_river',
+          shard_count: 2
+        })
+
+        client.add_tags_to_stream({
+          stream_name: 'deschutes_river',
+          tags: {'TagKey' => 'salmon'}
+        })
+      end
+      spans = harvest_span_events!
+      span = spans[1][1]
+
+      assert_equal 'Kinesis/add_tags_to_stream/deschutes_river', span[0]['name']
+      assert_equal 'aws_kinesis_data_streams', span[2]['cloud.platform']
+      assert_equal 'test-arn', span[2]['cloud.resource_id']
+    end
+  end
+
+  def test_list_tags_for_stream
+    client = create_client
+
+    NewRelic::Agent::Aws.stub(:create_arn, 'test-arn') do
+      in_transaction do |txn|
+        txn.stubs(:sampled?).returns(true)
+        client.create_stream({
+          stream_name: 'deschutes_river',
+          shard_count: 2
+        })
+
+        client.add_tags_to_stream({
+          stream_name: 'deschutes_river',
+          tags: {'TagKey' => 'salmon'}
+        })
+
+        client.list_tags_for_stream({
+          stream_name: 'deschutes_river'
+        })
+      end
+
+      spans = harvest_span_events!
+      span = spans[1][2]
+
+      assert_equal 'Kinesis/list_tags_for_stream/deschutes_river', span[0]['name']
+      assert_equal 'aws_kinesis_data_streams', span[2]['cloud.platform']
+      assert_equal 'test-arn', span[2]['cloud.resource_id']
+    end
+  end
+
+  def test_remove_tags_from_stream
+    client = create_client
+
+    NewRelic::Agent::Aws.stub(:create_arn, 'test-arn') do
+      in_transaction do |txn|
+        txn.stubs(:sampled?).returns(true)
+        client.create_stream({
+          stream_name: 'deschutes_river',
+          shard_count: 2
+        })
+
+        client.add_tags_to_stream({
+          stream_name: 'deschutes_river',
+          tags: {'TagKey' => 'salmon'}
+        })
+
+        client.remove_tags_from_stream({
+          stream_name: 'deschutes_river',
+          tag_keys: ['TagKey']
+        })
+      end
+
+      spans = harvest_span_events!
+      span = spans[1][2]
+
+      assert_equal 'Kinesis/remove_tags_from_stream/deschutes_river', span[0]['name']
+      assert_equal 'aws_kinesis_data_streams', span[2]['cloud.platform']
+      assert_equal 'test-arn', span[2]['cloud.resource_id']
+    end
+  end
+
+  def test_increase_stream_retention_period
+    client = create_client
+
+    NewRelic::Agent::Aws.stub(:create_arn, 'test-arn') do
+      in_transaction do |txn|
+        txn.stubs(:sampled?).returns(true)
+        client.create_stream({
+          stream_name: 'deschutes_river',
+          shard_count: 2
+        })
+
+        client.increase_stream_retention_period({
+          stream_name: 'deschutes_river',
+          retention_period_hours: 1
+        })
+      end
+
+      spans = harvest_span_events!
+      span = spans[1][1]
+
+      assert_equal 'Kinesis/increase_stream_retention_period/deschutes_river', span[0]['name']
+      assert_equal 'aws_kinesis_data_streams', span[2]['cloud.platform']
+      assert_equal 'test-arn', span[2]['cloud.resource_id']
+    end
+  end
+
+  def test_decrease_stream_retention_period
+    client = create_client
+
+    NewRelic::Agent::Aws.stub(:create_arn, 'test-arn') do
+      in_transaction do |txn|
+        txn.stubs(:sampled?).returns(true)
+        client.create_stream({
+          stream_name: 'deschutes_river',
+          shard_count: 2
+        })
+
+        client.decrease_stream_retention_period({
+          stream_name: 'deschutes_river',
+          retention_period_hours: 1
+        })
+      end
+
+      spans = harvest_span_events!
+      span = spans[1][1]
+
+      assert_equal 'Kinesis/decrease_stream_retention_period/deschutes_river', span[0]['name']
+      assert_equal 'aws_kinesis_data_streams', span[2]['cloud.platform']
+      assert_equal 'test-arn', span[2]['cloud.resource_id']
+    end
+  end
+
+  def test_put_record
+    client = create_client
+
+    NewRelic::Agent::Aws.stub(:create_arn, 'test-arn') do
+      in_transaction do |txn|
+        txn.stubs(:sampled?).returns(true)
+        client.create_stream({
+          stream_name: 'deschutes_river',
+          shard_count: 2
+        })
+
+        client.put_record({
+          stream_name: 'deschutes_river',
+          data: 'little lava lake',
+          partition_key: 'wickiup'
+        })
+      end
+
+      spans = harvest_span_events!
+      span = spans[1][1]
+
+      assert_metrics_recorded(['MessageBroker/Kinesis/Stream/Produce/Named/deschutes_river'])
+      assert_equal 'MessageBroker/Kinesis/Stream/Produce/Named/deschutes_river', span[0]['name']
+      assert_equal 'aws_kinesis_data_streams', span[2]['cloud.platform']
+      assert_equal 'test-arn', span[2]['cloud.resource_id']
+    end
+  end
+
+  def test_put_records
+    client = create_client
+
+    NewRelic::Agent::Aws.stub(:create_arn, 'test-arn') do
+      in_transaction do |txn|
+        txn.stubs(:sampled?).returns(true)
+        client.create_stream({
+          stream_name: 'deschutes_river',
+          shard_count: 2
+        })
+
+        client.put_records({
+          stream_name: 'deschutes_river',
+          records: [
+            {
+              data: 'spring chinook',
+              explicit_hash_key: 'HashKey',
+              partition_key: 'wickiup'
+            },
+            {
+              data: 'summer steelhead',
+              explicit_hash_key: 'HashKey',
+              partition_key: 'wickiup'
+            }
+          ]
+        })
+      end
+
+      spans = harvest_span_events!
+      span = spans[1][1]
+
+      assert_metrics_recorded(['MessageBroker/Kinesis/Stream/Produce/Named/deschutes_river'])
+      assert_equal 'MessageBroker/Kinesis/Stream/Produce/Named/deschutes_river', span[0]['name']
+      assert_equal 'aws_kinesis_data_streams', span[2]['cloud.platform']
+      assert_equal 'test-arn', span[2]['cloud.resource_id']
+    end
+  end
+
+  def test_get_records
+    client = create_client
+
+    NewRelic::Agent::Aws.stub(:create_arn, 'test-arn') do
+      in_transaction do |txn|
+        txn.stubs(:sampled?).returns(true)
+        client.create_stream({
+          stream_name: 'deschutes_river',
+          shard_count: 2
+        })
+
+        client.put_record({
+          stream_name: 'deschutes_river',
+          data: 'little lava lake',
+          partition_key: 'wickiup'
+        })
+
+        client.get_records({
+          shard_iterator: 'shard_iterator',
+          stream_arn: 'arn:aws:kinesis:us-east-1:123456789012:stream/deschutes_river'
+        })
+      end
+
+      spans = harvest_span_events!
+      span = spans[1][2]
+
+      assert_metrics_recorded(['MessageBroker/Kinesis/Stream/Consume/Named/deschutes_river'])
+      assert_equal 'MessageBroker/Kinesis/Stream/Consume/Named/deschutes_river', span[0]['name']
+      assert_equal 'aws_kinesis_data_streams', span[2]['cloud.platform']
+      assert_equal 'arn:aws:kinesis:us-east-1:123456789012:stream/deschutes_river', span[2]['cloud.resource_id']
+    end
+  end
+
+  def test_update_shard_count
+    client = create_client
+
+    NewRelic::Agent::Aws.stub(:create_arn, 'test-arn') do
+      in_transaction do |txn|
+        txn.stubs(:sampled?).returns(true)
+        client.create_stream({
+          stream_name: 'deschutes_river',
+          shard_count: 2
+        })
+
+        client.update_shard_count({
+          stream_name: 'deschutes_river',
+          target_shard_count: 4,
+          scaling_type: 'UNIFORM_SCALING'
+        })
+      end
+
+      spans = harvest_span_events!
+      span = spans[1][1]
+
+      assert_equal 'Kinesis/update_shard_count/deschutes_river', span[0]['name']
+      assert_equal 'aws_kinesis_data_streams', span[2]['cloud.platform']
+      assert_equal 'test-arn', span[2]['cloud.resource_id']
+    end
+  end
+
+  def test_split_shard
+    client = create_client
+
+    NewRelic::Agent::Aws.stub(:create_arn, 'test-arn') do
+      in_transaction do |txn|
+        txn.stubs(:sampled?).returns(true)
+        client.create_stream({
+          stream_name: 'deschutes_river',
+          shard_count: 2
+        })
+
+        client.split_shard({
+          stream_name: 'deschutes_river',
+          shard_to_split: 'shardId-000',
+          new_starting_hash_key: 'HashKey'
+        })
+      end
+
+      spans = harvest_span_events!
+      span = spans[1][1]
+
+      assert_equal 'Kinesis/split_shard/deschutes_river', span[0]['name']
+      assert_equal 'aws_kinesis_data_streams', span[2]['cloud.platform']
+      assert_equal 'test-arn', span[2]['cloud.resource_id']
+    end
+  end
+
+  def test_merge_shards
+    client = create_client
+
+    NewRelic::Agent::Aws.stub(:create_arn, 'test-arn') do
+      in_transaction do |txn|
+        txn.stubs(:sampled?).returns(true)
+        client.create_stream({
+          stream_name: 'deschutes_river',
+          shard_count: 2
+        })
+
+        client.merge_shards({
+          stream_name: 'deschutes_river',
+          shard_to_merge: 'shardId-000',
+          adjacent_shard_to_merge: 'shardId-001'
+        })
+      end
+
+      spans = harvest_span_events!
+      span = spans[1][1]
+
+      assert_equal 'Kinesis/merge_shards/deschutes_river', span[0]['name']
+      assert_equal 'aws_kinesis_data_streams', span[2]['cloud.platform']
+      assert_equal 'test-arn', span[2]['cloud.resource_id']
+    end
+  end
+
+  def test_get_shard_iterator
+    client = create_client
+
+    NewRelic::Agent::Aws.stub(:create_arn, 'test-arn') do
+      in_transaction do |txn|
+        txn.stubs(:sampled?).returns(true)
+        client.create_stream({
+          stream_name: 'deschutes_river',
+          shard_count: 2
+        })
+
+        client.get_shard_iterator({
+          stream_name: 'deschutes_river',
+          shard_id: 'shardId-000',
+          shard_iterator_type: 'TRIM_HORIZON'
+        })
+      end
+
+      spans = harvest_span_events!
+      span = spans[1][1]
+
+      assert_equal 'Kinesis/get_shard_iterator/deschutes_river', span[0]['name']
+      assert_equal 'aws_kinesis_data_streams', span[2]['cloud.platform']
+      assert_equal 'test-arn', span[2]['cloud.resource_id']
+    end
+  end
+
+  def test_enable_enhanced_monitoring
+    client = create_client
+
+    NewRelic::Agent::Aws.stub(:create_arn, 'test-arn') do
+      in_transaction do |txn|
+        txn.stubs(:sampled?).returns(true)
+        client.create_stream({
+          stream_name: 'deschutes_river',
+          shard_count: 2
+        })
+
+        client.enable_enhanced_monitoring({
+          stream_name: 'deschutes_river',
+          shard_level_metrics: ['ALL']
+        })
+      end
+
+      spans = harvest_span_events!
+      span = spans[1][1]
+
+      assert_equal 'Kinesis/enable_enhanced_monitoring/deschutes_river', span[0]['name']
+      assert_equal 'aws_kinesis_data_streams', span[2]['cloud.platform']
+      assert_equal 'test-arn', span[2]['cloud.resource_id']
+    end
+  end
+
+  def test_disable_enhanced_monitoring
+    client = create_client
+
+    NewRelic::Agent::Aws.stub(:create_arn, 'test-arn') do
+      in_transaction do |txn|
+        txn.stubs(:sampled?).returns(true)
+        client.create_stream({
+          stream_name: 'deschutes_river',
+          shard_count: 2
+        })
+
+        client.disable_enhanced_monitoring({
+          stream_name: 'deschutes_river',
+          shard_level_metrics: ['ALL']
+        })
+      end
+
+      spans = harvest_span_events!
+      span = spans[1][1]
+
+      assert_equal 'Kinesis/disable_enhanced_monitoring/deschutes_river', span[0]['name']
+      assert_equal 'aws_kinesis_data_streams', span[2]['cloud.platform']
+      assert_equal 'test-arn', span[2]['cloud.resource_id']
+    end
+  end
+
+  def test_describe_limits
+    client = create_client
+
+    NewRelic::Agent::Aws.stub(:create_arn, 'test-arn') do
+      in_transaction do |txn|
+        client.create_stream({
+          stream_name: 'deschutes_river',
+          shard_count: 2
+        })
+
+        client.describe_limits
+      end
+
+      spans = harvest_span_events!
+      span = spans[1][1]
+
+      assert_equal 'Kinesis/describe_limits', span[0]['name']
+      assert_equal 'aws_kinesis_data_streams', span[2]['cloud.platform']
+    end
+  end
+end