diff --git a/.changes/next-release/feature-AmazonCloudWatchEMFMetricPublisher-33792c9.json b/.changes/next-release/feature-AmazonCloudWatchEMFMetricPublisher-33792c9.json
new file mode 100644
index 000000000000..fb7ac8c8255f
--- /dev/null
+++ b/.changes/next-release/feature-AmazonCloudWatchEMFMetricPublisher-33792c9.json
@@ -0,0 +1,6 @@
+{
+    "type": "feature",
+    "category": "Amazon CloudWatch EMF Metric Publisher",
+    "contributor": "humanzz",
+    "description": "Add `PropertiesFactory` and `propertiesFactory` to `EmfMetricLoggingPublisher.Builder`, enabling users to enrich EMF records with custom key-value properties derived from the metric collection or ambient context, searchable in CloudWatch Logs Insights. See [#6595](https://github.com/aws/aws-sdk-java-v2/issues/6595)."
+}
diff --git a/metric-publishers/emf-metric-logging-publisher/src/main/java/software/amazon/awssdk/metrics/publishers/emf/EmfMetricLoggingPublisher.java b/metric-publishers/emf-metric-logging-publisher/src/main/java/software/amazon/awssdk/metrics/publishers/emf/EmfMetricLoggingPublisher.java
index 07bba8a97442..dbe858c23dac 100644
--- a/metric-publishers/emf-metric-logging-publisher/src/main/java/software/amazon/awssdk/metrics/publishers/emf/EmfMetricLoggingPublisher.java
+++ b/metric-publishers/emf-metric-logging-publisher/src/main/java/software/amazon/awssdk/metrics/publishers/emf/EmfMetricLoggingPublisher.java
@@ -82,6 +82,7 @@ private EmfMetricLoggingPublisher(Builder builder) {
             .dimensions(builder.dimensions)
             .metricLevel(builder.metricLevel)
             .metricCategories(builder.metricCategories)
+            .propertiesFactory(builder.propertiesFactory)
             .build();
 
         this.metricConverter = new MetricEmfConverter(config);
@@ -123,6 +124,7 @@ public static final class Builder {
         private Collection<SdkMetric<String>> dimensions;
         private Collection<MetricCategory> metricCategories;
         private MetricLevel metricLevel;
+        private PropertiesFactory propertiesFactory;
 
         private Builder() {
         }
@@ -217,6 +219,28 @@ public Builder metricLevel(MetricLevel metricLevel) {
         }
 
 
+        /**
+         * Configure a factory for custom properties to include in each EMF record.
+         * The factory is invoked on each {@link #publish(MetricCollection)} call with the
+         * {@link MetricCollection} being published, and the returned map entries are written
+         * as top-level key-value pairs in the EMF JSON output. These appear as searchable
+         * fields in CloudWatch Logs Insights.
+         *
+         * <p>Keys that collide with reserved EMF fields ({@code _aws}), configured
+         * dimension names, or reported metric names are silently skipped.
+         *
+         * <p>If this is not specified, no custom properties are added.
+         *
+         * @param propertiesFactory a factory returning a map of property names to values,
+         *                          or {@code null} to disable custom properties
+         * @return this builder
+         * @see PropertiesFactory
+         */
+        public Builder propertiesFactory(PropertiesFactory propertiesFactory) {
+            this.propertiesFactory = propertiesFactory;
+            return this;
+        }
+
         /**
          * Build a {@link EmfMetricLoggingPublisher} using the configuration currently configured on this publisher.
          */
diff --git a/metric-publishers/emf-metric-logging-publisher/src/main/java/software/amazon/awssdk/metrics/publishers/emf/PropertiesFactory.java b/metric-publishers/emf-metric-logging-publisher/src/main/java/software/amazon/awssdk/metrics/publishers/emf/PropertiesFactory.java
new file mode 100644
index 000000000000..9755406f57b9
--- /dev/null
+++ b/metric-publishers/emf-metric-logging-publisher/src/main/java/software/amazon/awssdk/metrics/publishers/emf/PropertiesFactory.java
@@ -0,0 +1,69 @@
+/*
+ * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License").
+ * You may not use this file except in compliance with the License.
+ * A copy of the License is located at
+ *
+ *  http://aws.amazon.com/apache2.0
+ *
+ * or in the "license" file accompanying this file. This file is distributed
+ * on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
+ * express or implied. See the License for the specific language governing
+ * permissions and limitations under the License.
+ */
+
+package software.amazon.awssdk.metrics.publishers.emf;
+
+import java.util.Map;
+import software.amazon.awssdk.annotations.SdkPublicApi;
+import software.amazon.awssdk.metrics.MetricCollection;
+
+/**
+ * A factory for producing custom properties to include in each EMF record.
+ *
+ * <p>Implementations receive the {@link MetricCollection} being published, allowing properties
+ * to be derived from the SDK metrics (e.g. service endpoint, request ID) or from ambient
+ * context (e.g. Lambda request ID, trace ID).
+ *
+ * <p>The returned map entries are written as top-level key-value pairs in the EMF JSON output,
+ * making them searchable in CloudWatch Logs Insights. Keys that collide with reserved EMF
+ * fields ({@code _aws}), dimension names, or metric names are silently skipped.
+ *
+ * <p>If the factory returns {@code null} or throws an exception, no custom properties are added
+ * and a warning is logged.
+ *
+ * <p>Example using ambient context:
+ * <pre>{@code
+ * EmfMetricLoggingPublisher.builder()
+ *     .propertiesFactory(metrics -> Collections.singletonMap("RequestId", requestId))
+ *     .build();
+ * }</pre>
+ *
+ * <p>Example using metric collection values:
+ * <pre>{@code
+ * EmfMetricLoggingPublisher.builder()
+ *     .propertiesFactory(metrics -> {
+ *         Map<String, String> props = new HashMap<>();
+ *         metrics.metricValues(CoreMetric.SERVICE_ENDPOINT)
+ *                .stream().findFirst()
+ *                .ifPresent(uri -> props.put("ServiceEndpoint", uri.toString()));
+ *         return props;
+ *     })
+ *     .build();
+ * }</pre>
+ *
+ * @see EmfMetricLoggingPublisher.Builder#propertiesFactory(PropertiesFactory)
+ */
+@FunctionalInterface
+@SdkPublicApi
+public interface PropertiesFactory {
+
+    /**
+     * Create a map of custom properties to include in the EMF record for the given metric collection.
+     *
+     * @param metricCollection the SDK metric collection being published
+     * @return a map of property names to string values, or {@code null} for no custom properties
+     */
+    Map<String, String> create(MetricCollection metricCollection);
+}
diff --git a/metric-publishers/emf-metric-logging-publisher/src/main/java/software/amazon/awssdk/metrics/publishers/emf/internal/EmfMetricConfiguration.java b/metric-publishers/emf-metric-logging-publisher/src/main/java/software/amazon/awssdk/metrics/publishers/emf/internal/EmfMetricConfiguration.java
index 739cf0b0e4ca..249da1178985 100644
--- a/metric-publishers/emf-metric-logging-publisher/src/main/java/software/amazon/awssdk/metrics/publishers/emf/internal/EmfMetricConfiguration.java
+++ b/metric-publishers/emf-metric-logging-publisher/src/main/java/software/amazon/awssdk/metrics/publishers/emf/internal/EmfMetricConfiguration.java
@@ -26,6 +26,7 @@
 import software.amazon.awssdk.metrics.MetricCategory;
 import software.amazon.awssdk.metrics.MetricLevel;
 import software.amazon.awssdk.metrics.SdkMetric;
+import software.amazon.awssdk.metrics.publishers.emf.PropertiesFactory;
 import software.amazon.awssdk.utils.Validate;
 import software.amazon.awssdk.utils.internal.SystemSettingUtils;
 
@@ -43,6 +44,7 @@ public final class EmfMetricConfiguration {
     private final Set<SdkMetric<String>> dimensions;
     private final Collection<MetricCategory> metricCategories;
     private final MetricLevel metricLevel;
+    private final PropertiesFactory propertiesFactory;
 
     private EmfMetricConfiguration(Builder builder) {
         this.namespace = builder.namespace == null ? DEFAULT_NAMESPACE : builder.namespace;
@@ -50,6 +52,9 @@ private EmfMetricConfiguration(Builder builder) {
         this.dimensions = builder.dimensions == null ? DEFAULT_DIMENSIONS : new HashSet<>(builder.dimensions);
         this.metricCategories = builder.metricCategories == null ? DEFAULT_CATEGORIES : new HashSet<>(builder.metricCategories);
         this.metricLevel = builder.metricLevel == null ? DEFAULT_METRIC_LEVEL : builder.metricLevel;
+        this.propertiesFactory = builder.propertiesFactory == null
+            ? mc -> Collections.emptyMap()
+            : builder.propertiesFactory;
     }
 
 
@@ -59,6 +64,7 @@ public static class Builder {
         private Collection<SdkMetric<String>> dimensions;
         private Collection<MetricCategory> metricCategories;
         private MetricLevel metricLevel;
+        private PropertiesFactory propertiesFactory;
 
         public Builder namespace(String namespace) {
             this.namespace = namespace;
@@ -85,6 +91,11 @@ public Builder metricLevel(MetricLevel metricLevel) {
             return this;
         }
 
+        public Builder propertiesFactory(PropertiesFactory propertiesFactory) {
+            this.propertiesFactory = propertiesFactory;
+            return this;
+        }
+
         public EmfMetricConfiguration build() {
             return new EmfMetricConfiguration(this);
         }
@@ -110,6 +121,10 @@ public MetricLevel metricLevel() {
         return metricLevel;
     }
 
+    public PropertiesFactory propertiesFactory() {
+        return propertiesFactory;
+    }
+
     private String resolveLogGroupName(Builder builder) {
         return builder.logGroupName != null ? builder.logGroupName :
                SystemSettingUtils.resolveEnvironmentVariable("AWS_LAMBDA_LOG_GROUP_NAME").orElse(null);
diff --git a/metric-publishers/emf-metric-logging-publisher/src/main/java/software/amazon/awssdk/metrics/publishers/emf/internal/MetricEmfConverter.java b/metric-publishers/emf-metric-logging-publisher/src/main/java/software/amazon/awssdk/metrics/publishers/emf/internal/MetricEmfConverter.java
index 3ab16d3b0878..6f9050f04db2 100644
--- a/metric-publishers/emf-metric-logging-publisher/src/main/java/software/amazon/awssdk/metrics/publishers/emf/internal/MetricEmfConverter.java
+++ b/metric-publishers/emf-metric-logging-publisher/src/main/java/software/amazon/awssdk/metrics/publishers/emf/internal/MetricEmfConverter.java
@@ -19,7 +19,9 @@
 import java.time.Clock;
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
@@ -31,6 +33,7 @@
 import software.amazon.awssdk.metrics.MetricCollection;
 import software.amazon.awssdk.metrics.MetricRecord;
 import software.amazon.awssdk.metrics.SdkMetric;
+import software.amazon.awssdk.metrics.publishers.emf.PropertiesFactory;
 import software.amazon.awssdk.protocols.jsoncore.JsonWriter;
 import software.amazon.awssdk.utils.Logger;
 import software.amazon.awssdk.utils.MetricValueNormalizer;
@@ -61,17 +64,21 @@ public class MetricEmfConverter {
      */
     private static final int MAX_METRIC_NUM = 100;
 
+    private static final String AWS_METADATA_KEY = "_aws";
+
     private static final Logger logger = Logger.loggerFor(MetricEmfConverter.class);
     private final List<String> dimensions = new ArrayList<>();
     private final EmfMetricConfiguration config;
     private final boolean metricCategoriesContainsAll;
     private final Clock clock;
+    private final PropertiesFactory propertiesFactory;
 
     @SdkTestInternalApi
     public MetricEmfConverter(EmfMetricConfiguration config, Clock clock) {
         this.config = config;
         this.clock = clock;
         this.metricCategoriesContainsAll = config.metricCategories().contains(MetricCategory.ALL);
+        this.propertiesFactory = config.propertiesFactory();
     }
 
     public MetricEmfConverter(EmfMetricConfiguration config) {
@@ -136,7 +143,18 @@ public List<String> convertMetricCollectionToEmf(MetricCollection metricCollecti
             }
         }
 
-        return createEmfStrings(aggregatedMetrics);
+        Map<String, String> properties = resolveProperties(metricCollection);
+        return createEmfStrings(aggregatedMetrics, properties);
+    }
+
+    private Map<String, String> resolveProperties(MetricCollection metricCollection) {
+        try {
+            Map<String, String> result = propertiesFactory.create(metricCollection);
+            return result == null ? Collections.emptyMap() : result;
+        } catch (Exception e) {
+            logger.warn(() -> "Properties factory threw an exception, publishing without custom properties", e);
+            return Collections.emptyMap();
+        }
     }
 
     /**
@@ -188,7 +206,8 @@ private void processAndWriteValue(JsonWriter jsonWriter, MetricRecord<?> mRecord
         }
     }
 
-    private List<String> createEmfStrings(Map<SdkMetric<?>, List<MetricRecord<?>>> aggregatedMetrics) {
+    private List<String> createEmfStrings(Map<SdkMetric<?>, List<MetricRecord<?>>> aggregatedMetrics,
+                                          Map<String, String> properties) {
         List<String> emfStrings = new ArrayList<>();
         Map<SdkMetric<?>, List<MetricRecord<?>>> currentMetricBatch = new HashMap<>();
 
@@ -204,26 +223,28 @@ private List<String> createEmfStrings(Map<SdkMetric<?>, List<MetricRecord<?>>> a
             }
 
             if (currentMetricBatch.size() == MAX_METRIC_NUM) {
-                emfStrings.add(createEmfString(currentMetricBatch));
+                emfStrings.add(createEmfString(currentMetricBatch, properties));
                 currentMetricBatch = new HashMap<>();
             }
 
             currentMetricBatch.put(metric, records);
         }
 
-        emfStrings.add(createEmfString(currentMetricBatch));
+        emfStrings.add(createEmfString(currentMetricBatch, properties));
 
         return emfStrings;
     }
 
 
-    private String createEmfString(Map<SdkMetric<?>, List<MetricRecord<?>>> metrics) {
+    private String createEmfString(Map<SdkMetric<?>, List<MetricRecord<?>>> metrics,
+                                    Map<String, String> properties) {
 
         JsonWriter jsonWriter = JsonWriter.create();
         jsonWriter.writeStartObject();
 
         writeAwsObject(jsonWriter, metrics.keySet());
         writeMetricValues(jsonWriter, metrics);
+        writeCustomProperties(jsonWriter, properties, metrics.keySet());
 
         jsonWriter.writeEndObject();
 
@@ -231,8 +252,26 @@ private String createEmfString(Map<SdkMetric<?>, List<MetricRecord<?>>> metrics)
 
     }
 
+    private void writeCustomProperties(JsonWriter jsonWriter, Map<String, String> properties,
+                                        Set<SdkMetric<?>> metrics) {
+        if (properties.isEmpty()) {
+            return;
+        }
+        Set<String> reservedKeys = new HashSet<>();
+        reservedKeys.add(AWS_METADATA_KEY);
+        for (SdkMetric<?> metric : metrics) {
+            reservedKeys.add(metric.name());
+        }
+        for (Map.Entry<String, String> entry : properties.entrySet()) {
+            if (!reservedKeys.contains(entry.getKey())) {
+                jsonWriter.writeFieldName(entry.getKey());
+                jsonWriter.writeValue(entry.getValue());
+            }
+        }
+    }
+
     private void writeAwsObject(JsonWriter jsonWriter, Set<SdkMetric<?>> metricNames) {
-        jsonWriter.writeFieldName("_aws");
+        jsonWriter.writeFieldName(AWS_METADATA_KEY);
         jsonWriter.writeStartObject();
 
         jsonWriter.writeFieldName("Timestamp");
diff --git a/metric-publishers/emf-metric-logging-publisher/src/test/java/software/amazon/awssdk/metrics/publishers/emf/EmfMetricLoggingPublisherTest.java b/metric-publishers/emf-metric-logging-publisher/src/test/java/software/amazon/awssdk/metrics/publishers/emf/EmfMetricLoggingPublisherTest.java
index f3da12abfb4c..659a2fddd969 100644
--- a/metric-publishers/emf-metric-logging-publisher/src/test/java/software/amazon/awssdk/metrics/publishers/emf/EmfMetricLoggingPublisherTest.java
+++ b/metric-publishers/emf-metric-logging-publisher/src/test/java/software/amazon/awssdk/metrics/publishers/emf/EmfMetricLoggingPublisherTest.java
@@ -18,6 +18,11 @@
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.logging.log4j.Level;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import software.amazon.awssdk.http.HttpMetric;
@@ -97,4 +102,96 @@ void Publish_multipleMetrics() {
         assertThat(loggedEvents()).hasSize(2);
     }
 
+    @Test
+    void publish_propertiesFactoryThrowsException_publishesWithoutCustomProperties() {
+        EmfMetricLoggingPublisher publisher = publisherBuilder
+            .logGroupName("/aws/lambda/emfMetricTest")
+            .propertiesFactory(mc -> { throw new RuntimeException("factory failed"); })
+            .build();
+
+        MetricCollector metricCollector = MetricCollector.create("test");
+        metricCollector.reportMetric(HttpMetric.AVAILABLE_CONCURRENCY, 5);
+        publisher.publish(metricCollector.collect());
+
+        // Should have: 1 warning about factory + 1 EMF info log
+        boolean hasWarning = loggedEvents().stream()
+            .anyMatch(e -> e.getLevel() == Level.WARN
+                && e.getMessage().getFormattedMessage().contains("Properties factory threw an exception"));
+        assertThat(hasWarning).isTrue();
+
+        boolean hasEmfOutput = loggedEvents().stream()
+            .anyMatch(e -> e.getLevel() == Level.INFO
+                && e.getMessage().getFormattedMessage().contains("\"_aws\":{"));
+        assertThat(hasEmfOutput).isTrue();
+
+        // EMF output should not contain any custom properties
+        String emfLog = loggedEvents().stream()
+            .filter(e -> e.getLevel() == Level.INFO
+                && e.getMessage().getFormattedMessage().contains("\"_aws\":{"))
+            .findFirst().get().getMessage().getFormattedMessage();
+        assertThat(emfLog).contains("\"AvailableConcurrency\":5");
+    }
+
+    @Test
+    void publish_propertiesFactoryReturnsNull_publishesWithoutCustomProperties() {
+        EmfMetricLoggingPublisher publisher = publisherBuilder
+            .logGroupName("/aws/lambda/emfMetricTest")
+            .propertiesFactory(mc -> null)
+            .build();
+
+        MetricCollector metricCollector = MetricCollector.create("test");
+        metricCollector.reportMetric(HttpMetric.AVAILABLE_CONCURRENCY, 5);
+        publisher.publish(metricCollector.collect());
+
+        // Should have EMF output without custom properties
+        boolean hasEmfOutput = loggedEvents().stream()
+            .anyMatch(e -> e.getLevel() == Level.INFO
+                && e.getMessage().getFormattedMessage().contains("\"_aws\":{"));
+        assertThat(hasEmfOutput).isTrue();
+
+        String emfLog = loggedEvents().stream()
+            .filter(e -> e.getLevel() == Level.INFO
+                && e.getMessage().getFormattedMessage().contains("\"_aws\":{"))
+            .findFirst().get().getMessage().getFormattedMessage();
+        assertThat(emfLog).contains("\"AvailableConcurrency\":5");
+        // No warning should be logged for null return
+        boolean hasWarning = loggedEvents().stream()
+            .anyMatch(e -> e.getLevel() == Level.WARN);
+        assertThat(hasWarning).isFalse();
+    }
+
+    @Test
+    void publish_statefulFactory_eachPublishUsesCurrentMap() {
+        AtomicInteger counter = new AtomicInteger(0);
+        EmfMetricLoggingPublisher publisher = publisherBuilder
+            .logGroupName("/aws/lambda/emfMetricTest")
+            .propertiesFactory(mc -> {
+                int count = counter.incrementAndGet();
+                Map<String, String> map = new HashMap<String, String>();
+                map.put("InvocationCount", String.valueOf(count));
+                return map;
+            })
+            .build();
+
+        // First publish
+        MetricCollector mc1 = MetricCollector.create("test1");
+        mc1.reportMetric(HttpMetric.AVAILABLE_CONCURRENCY, 5);
+        publisher.publish(mc1.collect());
+
+        // Second publish
+        MetricCollector mc2 = MetricCollector.create("test2");
+        mc2.reportMetric(HttpMetric.AVAILABLE_CONCURRENCY, 10);
+        publisher.publish(mc2.collect());
+
+        // Collect all EMF info logs
+        List<String> emfLogs = loggedEvents().stream()
+            .filter(e -> e.getLevel() == Level.INFO
+                && e.getMessage().getFormattedMessage().contains("\"_aws\":{"))
+            .map(e -> e.getMessage().getFormattedMessage())
+            .collect(java.util.stream.Collectors.toList());
+
+        assertThat(emfLogs).hasSize(2);
+        assertThat(emfLogs.get(0)).contains("\"InvocationCount\":\"1\"");
+        assertThat(emfLogs.get(1)).contains("\"InvocationCount\":\"2\"");
+    }
 }
diff --git a/metric-publishers/emf-metric-logging-publisher/src/test/java/software/amazon/awssdk/metrics/publishers/emf/internal/MetricEmfConverterTest.java b/metric-publishers/emf-metric-logging-publisher/src/test/java/software/amazon/awssdk/metrics/publishers/emf/internal/MetricEmfConverterTest.java
index a01af0f96320..a33522af594e 100644
--- a/metric-publishers/emf-metric-logging-publisher/src/test/java/software/amazon/awssdk/metrics/publishers/emf/internal/MetricEmfConverterTest.java
+++ b/metric-publishers/emf-metric-logging-publisher/src/test/java/software/amazon/awssdk/metrics/publishers/emf/internal/MetricEmfConverterTest.java
@@ -26,7 +26,9 @@
 import java.time.Clock;
 import java.time.Instant;
 import java.time.ZoneOffset;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -250,4 +252,135 @@ void ConvertMetricCollectionToEMF_longValueShouldSucceed() {
                                          + "\"CloudWatchMetrics\":[{\"Namespace\":\"AwsSdk/JavaSdk2\",\"Dimensions\":[[]],"
                                          + "\"Metrics\":[{\"Name\":\"TestMetric\"}]}]},\"TestMetric\":42}");
     }
-}
\ No newline at end of file
+
+    @Test
+    void convertMetricCollectionToEmf_withCustomProperties_propertiesAppearInOutput() {
+        Map<String, String> properties = new HashMap<String, String>();
+        properties.put("RequestId", "abc-123");
+        properties.put("FunctionName", "myLambda");
+
+        MetricEmfConverter converter = converterWithProperties(properties);
+        MetricCollector metricCollector = MetricCollector.create("test");
+        metricCollector.reportMetric(HttpMetric.AVAILABLE_CONCURRENCY, 5);
+
+        List<String> emfLogs = converter.convertMetricCollectionToEmf(metricCollector.collect());
+
+        assertThat(emfLogs).hasSize(1);
+        String emfLog = emfLogs.get(0);
+        assertThat(emfLog).contains("\"RequestId\":\"abc-123\"");
+        assertThat(emfLog).contains("\"FunctionName\":\"myLambda\"");
+        assertThat(emfLog).contains("\"AvailableConcurrency\":5");
+        assertThat(emfLog).contains("\"_aws\":{");
+    }
+
+    @Test
+    void convertMetricCollectionToEmf_noProperties_identicalToCurrentBehavior() {
+        MetricCollector metricCollector = MetricCollector.create("test");
+        metricCollector.reportMetric(HttpMetric.AVAILABLE_CONCURRENCY, 5);
+
+        List<String> emfLogs = metricEmfConverterDefault.convertMetricCollectionToEmf(metricCollector.collect());
+
+        assertThat(emfLogs).containsOnly(
+            "{\"_aws\":{\"Timestamp\":12345678,\"LogGroupName\":\"my_log_group_name\","
+            + "\"CloudWatchMetrics\":[{\"Namespace\":\"AwsSdk/JavaSdk2\",\"Dimensions\":[[]],"
+            + "\"Metrics\":[{\"Name\":\"AvailableConcurrency\"}]}]},\"AvailableConcurrency\":5}");
+    }
+
+    @Test
+    void convertMetricCollectionToEmf_emptyProperties_noExtraKeys() {
+        MetricEmfConverter converter = converterWithProperties(new HashMap<String, String>());
+        MetricCollector metricCollector = MetricCollector.create("test");
+        metricCollector.reportMetric(HttpMetric.AVAILABLE_CONCURRENCY, 5);
+
+        List<String> emfLogs = converter.convertMetricCollectionToEmf(metricCollector.collect());
+
+        assertThat(emfLogs).hasSize(1);
+        String emfLog = emfLogs.get(0);
+        assertThat(emfLog).doesNotContain("\"RequestId\"");
+        assertThat(emfLog).doesNotContain("\"FunctionName\"");
+    }
+
+    @Test
+    void convertMetricCollectionToEmf_propertyKeyMatchesAwsKey_awsObjectPreserved() {
+        Map<String, String> properties = new HashMap<String, String>();
+        properties.put("_aws", "should-be-overwritten");
+
+        MetricEmfConverter converter = converterWithProperties(properties);
+        MetricCollector metricCollector = MetricCollector.create("test");
+        metricCollector.reportMetric(HttpMetric.AVAILABLE_CONCURRENCY, 5);
+
+        List<String> emfLogs = converter.convertMetricCollectionToEmf(metricCollector.collect());
+
+        assertThat(emfLogs).hasSize(1);
+        String emfLog = emfLogs.get(0);
+        assertThat(emfLog).doesNotContain("should-be-overwritten");
+        assertThat(emfLog).contains("\"_aws\":{\"Timestamp\":");
+        assertThat(emfLog).contains("\"CloudWatchMetrics\":");
+    }
+
+    @Test
+    void convertMetricCollectionToEmf_propertyKeyMatchesDimensionName_dimensionPreserved() {
+        Map<String, String> properties = new HashMap<String, String>();
+        properties.put("OperationName", "should-be-overwritten");
+
+        MetricEmfConverter converter = converterWithProperties(properties);
+        MetricCollector metricCollector = MetricCollector.create("test");
+        metricCollector.reportMetric(CoreMetric.OPERATION_NAME, "GetObject");
+        metricCollector.reportMetric(HttpMetric.AVAILABLE_CONCURRENCY, 5);
+
+        List<String> emfLogs = converter.convertMetricCollectionToEmf(metricCollector.collect());
+
+        assertThat(emfLogs).hasSize(1);
+        String emfLog = emfLogs.get(0);
+        assertThat(emfLog).doesNotContain("should-be-overwritten");
+        assertThat(emfLog).contains("\"OperationName\":\"GetObject\"");
+    }
+
+    @Test
+    void convertMetricCollectionToEmf_propertyKeyMatchesMetricName_metricPreserved() {
+        Map<String, String> properties = new HashMap<String, String>();
+        properties.put("AvailableConcurrency", "should-be-overwritten");
+
+        MetricEmfConverter converter = converterWithProperties(properties);
+        MetricCollector metricCollector = MetricCollector.create("test");
+        metricCollector.reportMetric(HttpMetric.AVAILABLE_CONCURRENCY, 5);
+
+        List<String> emfLogs = converter.convertMetricCollectionToEmf(metricCollector.collect());
+
+        assertThat(emfLogs).hasSize(1);
+        String emfLog = emfLogs.get(0);
+        assertThat(emfLog).doesNotContain("should-be-overwritten");
+        assertThat(emfLog).contains("\"AvailableConcurrency\":5");
+    }
+
+    @Test
+    void convertMetricCollectionToEmf_batchedRecords_allContainCustomProperties() {
+        Map<String, String> properties = new HashMap<String, String>();
+        properties.put("RequestId", "batch-test-123");
+        properties.put("TraceId", "trace-abc");
+
+        MetricEmfConverter converter = converterWithProperties(properties);
+        MetricCollector metricCollector = MetricCollector.create("test");
+        for (int i = 0; i < 220; i++) {
+            metricCollector.reportMetric(
+                SdkMetric.create("cp_batch_metric_" + i, Integer.class, MetricLevel.INFO, MetricCategory.CORE), i);
+        }
+
+        List<String> emfLogs = converter.convertMetricCollectionToEmf(metricCollector.collect());
+
+        assertThat(emfLogs).hasSize(3);
+        for (String emfLog : emfLogs) {
+            assertThat(emfLog).contains("\"RequestId\":\"batch-test-123\"");
+            assertThat(emfLog).contains("\"TraceId\":\"trace-abc\"");
+            assertThat(emfLog).contains("\"_aws\":{");
+        }
+    }
+
+    private MetricEmfConverter converterWithProperties(Map<String, String> properties) {
+        EmfMetricConfiguration config = new EmfMetricConfiguration.Builder()
+            .logGroupName("my_log_group_name")
+            .propertiesFactory(mc -> properties)
+            .build();
+        return new MetricEmfConverter(config, fixedClock);
+    }
+}