diff --git a/aws-logs-loggroup/src/main/java/software/amazon/logs/loggroup/Configuration.java b/aws-logs-loggroup/src/main/java/software/amazon/logs/loggroup/Configuration.java
index 2c95be4..6f9900c 100644
--- a/aws-logs-loggroup/src/main/java/software/amazon/logs/loggroup/Configuration.java
+++ b/aws-logs-loggroup/src/main/java/software/amazon/logs/loggroup/Configuration.java
@@ -2,8 +2,10 @@
 
 import org.json.JSONObject;
 import org.json.JSONTokener;
+import software.amazon.awssdk.utils.CollectionUtils;
 
 import java.util.Map;
+import java.util.stream.Collectors;
 
 class Configuration extends BaseConfiguration {
 
@@ -16,6 +18,12 @@ public JSONObject resourceSchemaJSONObject() {
     }
 
     public Map<String, String> resourceDefinedTags(final ResourceModel resourceModel) {
-        return null;
+        if (CollectionUtils.isNullOrEmpty(resourceModel.getTags())) {
+            return null;
+        }
+
+        return resourceModel.getTags()
+                .stream()
+                .collect(Collectors.toMap(Tag::getKey, Tag::getValue, (value1, value2) -> value2));
     }
 }
diff --git a/aws-logs-loggroup/src/main/java/software/amazon/logs/loggroup/CreateHandler.java b/aws-logs-loggroup/src/main/java/software/amazon/logs/loggroup/CreateHandler.java
index 5fcbc94..79206ab 100644
--- a/aws-logs-loggroup/src/main/java/software/amazon/logs/loggroup/CreateHandler.java
+++ b/aws-logs-loggroup/src/main/java/software/amazon/logs/loggroup/CreateHandler.java
@@ -26,7 +26,7 @@ public ProgressEvent<ResourceModel, CallbackContext> handleRequest(
         final ResourceModel model = request.getDesiredResourceState();
 
         try {
-            proxy.injectCredentialsAndInvokeV2(Translator.translateToCreateRequest(model),
+            proxy.injectCredentialsAndInvokeV2(Translator.translateToCreateRequest(model, request.getDesiredResourceTags()),
                 ClientBuilder.getClient()::createLogGroup);
         } catch (final ResourceAlreadyExistsException e) {
             throw new CfnAlreadyExistsException(ResourceModel.TYPE_NAME,
diff --git a/aws-logs-loggroup/src/main/java/software/amazon/logs/loggroup/Translator.java b/aws-logs-loggroup/src/main/java/software/amazon/logs/loggroup/Translator.java
index 1064b3c..9dff99d 100644
--- a/aws-logs-loggroup/src/main/java/software/amazon/logs/loggroup/Translator.java
+++ b/aws-logs-loggroup/src/main/java/software/amazon/logs/loggroup/Translator.java
@@ -12,6 +12,7 @@
 import software.amazon.awssdk.services.cloudwatchlogs.model.AssociateKmsKeyRequest;
 import software.amazon.awssdk.services.cloudwatchlogs.model.TagLogGroupRequest;
 import software.amazon.awssdk.services.cloudwatchlogs.model.UntagLogGroupRequest;
+import software.amazon.awssdk.utils.CollectionUtils;
 
 import java.util.Collection;
 import java.util.Collections;
@@ -45,11 +46,11 @@ static DeleteLogGroupRequest translateToDeleteRequest(final ResourceModel model)
                 .build();
     }
 
-    static CreateLogGroupRequest translateToCreateRequest(final ResourceModel model) {
+    static CreateLogGroupRequest translateToCreateRequest(final ResourceModel model, final Map<String, String> tags) {
         return CreateLogGroupRequest.builder()
                 .logGroupName(model.getLogGroupName())
                 .kmsKeyId(model.getKmsKeyId())
-                .tags(translateTagsToSdk(model.getTags()))
+                .tags(tags)
                 .build();
     }
 
@@ -85,10 +86,10 @@ static ListTagsLogGroupRequest translateToListTagsLogGroupRequest(final String l
                 .build();
     }
 
-    static TagLogGroupRequest translateToTagLogGroupRequest(final String logGroupName, final Set<Tag> tags) {
+    static TagLogGroupRequest translateToTagLogGroupRequest(final String logGroupName, final Map<String, String> tags) {
         return TagLogGroupRequest.builder()
                 .logGroupName(logGroupName)
-                .tags(translateTagsToSdk(tags))
+                .tags(tags)
                 .build();
     }
 
@@ -163,14 +164,14 @@ static String buildResourceDoesNotExistErrorMessage(final String resourceIdentif
     }
 
     static Map<String, String> translateTagsToSdk(final Set<Tag> tags) {
-        if (tags == null || tags.isEmpty()) {
+        if (CollectionUtils.isNullOrEmpty(tags)) {
             return null;
         }
         return tags.stream().collect(Collectors.toMap(Tag::getKey, Tag::getValue));
     }
 
     static Set<Tag> translateSdkToTags(final Map<String, String> tags) {
-        if (tags == null || tags.isEmpty()) {
+        if (CollectionUtils.isNullOrEmpty(tags)) {
             return null;
         }
         return tags.entrySet().stream().map(tag -> new Tag(tag.getKey(), tag.getValue()))
diff --git a/aws-logs-loggroup/src/main/java/software/amazon/logs/loggroup/UpdateHandler.java b/aws-logs-loggroup/src/main/java/software/amazon/logs/loggroup/UpdateHandler.java
index 14f33d7..0919385 100644
--- a/aws-logs-loggroup/src/main/java/software/amazon/logs/loggroup/UpdateHandler.java
+++ b/aws-logs-loggroup/src/main/java/software/amazon/logs/loggroup/UpdateHandler.java
@@ -1,6 +1,7 @@
 package software.amazon.logs.loggroup;
 
-import com.google.common.collect.Sets;
+import com.google.common.collect.MapDifference;
+import com.google.common.collect.Maps;
 import software.amazon.cloudformation.exceptions.CfnInternalFailureException;
 import software.amazon.cloudformation.exceptions.CfnResourceConflictException;
 import software.amazon.cloudformation.exceptions.CfnServiceInternalErrorException;
@@ -17,12 +18,14 @@
 import software.amazon.awssdk.services.cloudwatchlogs.model.ResourceNotFoundException;
 import software.amazon.awssdk.services.cloudwatchlogs.model.ServiceUnavailableException;
 
-import java.util.HashSet;
+import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
-import java.util.Set;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 public class UpdateHandler extends BaseHandler<CallbackContext> {
 
@@ -35,9 +38,12 @@ public ProgressEvent<ResourceModel, CallbackContext> handleRequest(
 
         final ResourceModel model = request.getDesiredResourceState();
         final ResourceModel previousModel = request.getPreviousResourceState();
+        final Map<String, String> tags = request.getDesiredResourceTags();
+        final Map<String, String> previousTags = request.getPreviousResourceTags();
+
         final boolean retentionChanged = ! retentionUnchanged(previousModel, model);
         final boolean kmsKeyChanged = ! kmsKeyUnchanged(previousModel, model);
-        final boolean tagsChanged = ! tagsUnchanged(previousModel, model);
+        final boolean tagsChanged = ! tagsUnchanged(previousTags, tags);
         if (retentionChanged && model.getRetentionInDays() == null) {
             deleteRetentionPolicy(proxy, request, logger);
         } else if (retentionChanged){
@@ -54,7 +60,7 @@ public ProgressEvent<ResourceModel, CallbackContext> handleRequest(
         }
 
         if (tagsChanged) {
-            updateTags(proxy, previousModel, model, logger);
+            updateTags(proxy, model, previousTags, tags, logger);
         }
 
         return ProgressEvent.defaultSuccessHandler(model);
@@ -159,16 +165,21 @@ private void associateKmsKey(final AmazonWebServicesClientProxy proxy,
     }
 
     private void updateTags(final AmazonWebServicesClientProxy proxy,
-                            final ResourceModel previousModel,
                             final ResourceModel model,
+                            final Map<String, String> previousTags,
+                            final Map<String, String> tags,
                             final Logger logger) {
-        final Set<Tag> previousTags = Optional.ofNullable(previousModel).map(ResourceModel::getTags).orElse(new HashSet<>());
-        final Set<Tag> newTags = Optional.ofNullable(model.getTags()).orElse(new HashSet<>());
-        final Set<Tag> tagsToRemove = Sets.difference(previousTags, newTags);
-        final Set<Tag> tagsToAdd = Sets.difference(newTags, previousTags);
+        MapDifference<String, String> tagsDifference = Maps.difference(Optional.ofNullable(previousTags).orElse(new HashMap<>()),
+                Optional.ofNullable(tags).orElse(new HashMap<>()));
+        final Map<String, String> tagsToRemove = tagsDifference.entriesOnlyOnLeft();
+        final Map<String, String> tagsToAdd = tagsDifference.entriesOnlyOnRight();
+        final Map<String, String> tagsToDiffer = tagsDifference.entriesDiffering().entrySet().stream()
+                .collect(Collectors.toMap(Map.Entry::getKey, tag -> tag.getValue().rightValue()));
+        final Map<String, String> tagsToUpdate = Stream.concat(tagsToAdd.entrySet().stream(), tagsToDiffer.entrySet().stream())
+                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
         try {
             if (!tagsToRemove.isEmpty()) {
-                final List<String> tagKeys = tagsToRemove.stream().map(Tag::getKey).collect(Collectors.toList());
+                final List<String> tagKeys = new ArrayList<>(tagsToRemove.keySet());
                 proxy.injectCredentialsAndInvokeV2(Translator.translateToUntagLogGroupRequest(model.getLogGroupName(), tagKeys),
                         ClientBuilder.getClient()::untagLogGroup);
 
@@ -177,13 +188,13 @@ private void updateTags(final AmazonWebServicesClientProxy proxy,
                                 ResourceModel.TYPE_NAME, model.getLogGroupName(), tagKeys);
                 logger.log(message);
             }
-            if(!tagsToAdd.isEmpty()) {
-                proxy.injectCredentialsAndInvokeV2(Translator.translateToTagLogGroupRequest(model.getLogGroupName(), tagsToAdd),
+            if(!tagsToUpdate.isEmpty()) {
+                proxy.injectCredentialsAndInvokeV2(Translator.translateToTagLogGroupRequest(model.getLogGroupName(), tagsToUpdate),
                         ClientBuilder.getClient()::tagLogGroup);
 
                 final String message =
                         String.format("%s [%s] successfully added tags: [%s]",
-                                ResourceModel.TYPE_NAME, model.getLogGroupName(), tagsToAdd);
+                                ResourceModel.TYPE_NAME, model.getLogGroupName(), tagsToUpdate);
                 logger.log(message);
             }
         } catch (final ResourceNotFoundException e) {
@@ -207,10 +218,10 @@ private static boolean kmsKeyUnchanged(final ResourceModel previousModel, final
         return (previousModel != null && Objects.equals(model.getKmsKeyId(), previousModel.getKmsKeyId()));
     }
 
-    private static boolean tagsUnchanged(final ResourceModel previousModel, final ResourceModel model) {
-        if (previousModel == null && model.getTags() == null) {
+    private static boolean tagsUnchanged(final Map<String, String> previousTags, final Map<String, String> tags) {
+        if (previousTags == null && tags == null) {
             return true;
         }
-        return (previousModel != null && Objects.equals(model.getTags(), previousModel.getTags()));
+        return (previousTags != null && Objects.equals(previousTags, tags));
     }
 }
diff --git a/aws-logs-loggroup/src/test/java/software/amazon/logs/loggroup/ConfigurationTest.java b/aws-logs-loggroup/src/test/java/software/amazon/logs/loggroup/ConfigurationTest.java
new file mode 100644
index 0000000..c4fe0ed
--- /dev/null
+++ b/aws-logs-loggroup/src/test/java/software/amazon/logs/loggroup/ConfigurationTest.java
@@ -0,0 +1,28 @@
+package software.amazon.logs.loggroup;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class ConfigurationTest {
+
+    @Test
+    public void testResourceDefinedTags_MergeDuplicateKeys() {
+        final Set<Tag> tags = new HashSet<>(Arrays.asList(
+                Tag.builder().key("key-1").value("value-1").build(),
+                Tag.builder().key("key-1").value("value-2").build()
+        ));
+        final ResourceModel model = ResourceModel.builder()
+                .tags(tags)
+                .build();
+
+        final Configuration configuration = new Configuration();
+
+        assertThat(configuration.resourceDefinedTags(model)).isEqualTo(Collections.singletonMap("key-1", "value-2"));
+    }
+}
diff --git a/aws-logs-loggroup/src/test/java/software/amazon/logs/loggroup/CreateHandlerTest.java b/aws-logs-loggroup/src/test/java/software/amazon/logs/loggroup/CreateHandlerTest.java
index db1c92a..eb4c942 100644
--- a/aws-logs-loggroup/src/test/java/software/amazon/logs/loggroup/CreateHandlerTest.java
+++ b/aws-logs-loggroup/src/test/java/software/amazon/logs/loggroup/CreateHandlerTest.java
@@ -1,5 +1,9 @@
 package software.amazon.logs.loggroup;
 
+import org.mockito.ArgumentCaptor;
+import software.amazon.awssdk.services.cloudwatchlogs.model.CloudWatchLogsRequest;
+import software.amazon.awssdk.services.cloudwatchlogs.model.CreateLogGroupRequest;
+import software.amazon.awssdk.services.cloudwatchlogs.model.PutRetentionPolicyRequest;
 import software.amazon.cloudformation.exceptions.CfnAlreadyExistsException;
 import software.amazon.cloudformation.proxy.AmazonWebServicesClientProxy;
 import software.amazon.cloudformation.proxy.Logger;
@@ -18,12 +22,9 @@
 import software.amazon.awssdk.services.cloudwatchlogs.model.PutRetentionPolicyResponse;
 import software.amazon.awssdk.services.cloudwatchlogs.model.ResourceAlreadyExistsException;
 
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.Map;
-import java.util.Set;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -58,10 +59,10 @@ public void handleRequest_Success() {
                 .retentionInDays(1)
                 .kmsKeyId("arn:aws:kms:us-east-1:$123456789012:key/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")
                 .build();
-        final Set<Tag> tags = new HashSet<>(Arrays.asList(
-                Tag.builder().key("key-1").value("value-1").build(),
-                Tag.builder().key("key-2").value("value-2").build()
-        ));
+        final Map<String, String> tags = new HashMap<String, String>() {{
+            put("key-1", "value-1");
+            put("key-2", "value-2");
+        }};
 
         doReturn(describeResponseInitial, createLogGroupResponse, putRetentionPolicyResponse)
             .when(proxy)
@@ -74,22 +75,33 @@ public void handleRequest_Success() {
                 .logGroupName("LogGroup")
                 .retentionInDays(1)
                 .kmsKeyId("arn:aws:kms:us-east-1:$123456789012:key/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")
-                .tags(tags)
                 .build();
 
         final ResourceHandlerRequest<ResourceModel> request = ResourceHandlerRequest.<ResourceModel>builder()
-            .desiredResourceState(model)
-            .build();
+                .desiredResourceState(model)
+                .desiredResourceTags(tags)
+                .build();
 
         final ProgressEvent<ResourceModel, CallbackContext> response = handler.handleRequest(proxy, request, null, logger);
 
+        ArgumentCaptor<CloudWatchLogsRequest> requests = ArgumentCaptor.forClass(CloudWatchLogsRequest.class);
+        verify(proxy, times(2)).injectCredentialsAndInvokeV2(requests.capture(), any());
+        assertThat(requests.getAllValues().get(0)).isEqualTo(CreateLogGroupRequest.builder()
+                .logGroupName("LogGroup")
+                .kmsKeyId("arn:aws:kms:us-east-1:$123456789012:key/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")
+                .tags(tags)
+                .build());
+        assertThat(requests.getAllValues().get(1)).isEqualTo(PutRetentionPolicyRequest.builder()
+                .logGroupName("LogGroup")
+                .retentionInDays(1)
+                .build());
+
         assertThat(response).isNotNull();
         assertThat(response.getStatus()).isEqualTo(OperationStatus.SUCCESS);
         assertThat(response.getCallbackContext()).isNull();
         assertThat(response.getCallbackDelaySeconds()).isEqualTo(0);
         assertThat(response.getResourceModels()).isNull();
         assertThat(response.getResourceModel()).isEqualToComparingOnlyGivenFields(logGroup);
-        assertThat(response.getResourceModel().getTags()).isEqualTo(tags);
         assertThat(response.getMessage()).isNull();
         assertThat(response.getErrorCode()).isNull();
     }
diff --git a/aws-logs-loggroup/src/test/java/software/amazon/logs/loggroup/TranslatorTest.java b/aws-logs-loggroup/src/test/java/software/amazon/logs/loggroup/TranslatorTest.java
index cc815ac..c76184c 100644
--- a/aws-logs-loggroup/src/test/java/software/amazon/logs/loggroup/TranslatorTest.java
+++ b/aws-logs-loggroup/src/test/java/software/amazon/logs/loggroup/TranslatorTest.java
@@ -79,7 +79,7 @@ public void testTranslateToCreate() {
                 .kmsKeyId(RESOURCE_MODEL.getKmsKeyId())
                 .tags(MAP_TAGS)
                 .build();
-        assertThat(Translator.translateToCreateRequest(RESOURCE_MODEL)).isEqualToComparingFieldByField(request);
+        assertThat(Translator.translateToCreateRequest(RESOURCE_MODEL, MAP_TAGS)).isEqualToComparingFieldByField(request);
     }
 
     @Test
@@ -141,7 +141,7 @@ public void testTranslateToTagLogGroupRequest() {
                 .tags(MAP_TAGS)
                 .build();
 
-        assertThat(Translator.translateToTagLogGroupRequest(RESOURCE_MODEL.getLogGroupName(), SET_TAGS))
+        assertThat(Translator.translateToTagLogGroupRequest(RESOURCE_MODEL.getLogGroupName(), MAP_TAGS))
                 .isEqualToComparingFieldByField(request);
     }
 
diff --git a/aws-logs-loggroup/src/test/java/software/amazon/logs/loggroup/UpdateHandlerTest.java b/aws-logs-loggroup/src/test/java/software/amazon/logs/loggroup/UpdateHandlerTest.java
index 708b0bb..018f6ad 100644
--- a/aws-logs-loggroup/src/test/java/software/amazon/logs/loggroup/UpdateHandlerTest.java
+++ b/aws-logs-loggroup/src/test/java/software/amazon/logs/loggroup/UpdateHandlerTest.java
@@ -1,6 +1,6 @@
 package software.amazon.logs.loggroup;
 
-import com.google.common.collect.Sets;
+import com.google.common.collect.Maps;
 import software.amazon.awssdk.services.cloudwatchlogs.model.CloudWatchLogsRequest;
 import software.amazon.awssdk.services.cloudwatchlogs.model.DeleteRetentionPolicyResponse;
 import software.amazon.awssdk.services.cloudwatchlogs.model.DescribeLogGroupsResponse;
@@ -26,12 +26,13 @@
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
 
-import java.util.Arrays;
+import java.util.ArrayList;
 import java.util.Collections;
-import java.util.HashSet;
+import java.util.HashMap;
 import java.util.List;
-import java.util.Set;
+import java.util.Map;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.jupiter.api.Assertions.assertThrows;
@@ -67,10 +68,10 @@ public void handleRequest_Success() {
                 .retentionInDays(1)
                 .kmsKeyId("arn:aws:kms:us-east-1:$123456789012:key/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")
                 .build();
-        final Set<Tag> tags = new HashSet<>(Arrays.asList(
-                Tag.builder().key("key-1").value("value-1").build(),
-                Tag.builder().key("key-2").value("value-2").build()
-        ));
+        final Map<String, String> tags = new HashMap<String, String>() {{
+            put("key-1", "value-1");
+            put("key-2", "value-2");
+        }};
         final DescribeLogGroupsResponse describeResponse = DescribeLogGroupsResponse.builder()
                 .logGroups(Collections.singletonList(logGroup))
                 .build();
@@ -86,11 +87,11 @@ public void handleRequest_Success() {
                 .logGroupName("LogGroup")
                 .retentionInDays(1)
                 .kmsKeyId("arn:aws:kms:us-east-1:$123456789012:key/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")
-                .tags(tags)
                 .build();
 
         final ResourceHandlerRequest<ResourceModel> request = ResourceHandlerRequest.<ResourceModel>builder()
             .desiredResourceState(model)
+                .desiredResourceTags(tags)
             .build();
 
         final ProgressEvent<ResourceModel, CallbackContext> response = handler.handleRequest(proxy, request, null, logger);
@@ -101,7 +102,6 @@ public void handleRequest_Success() {
         assertThat(response.getCallbackDelaySeconds()).isEqualTo(0);
         assertThat(response.getResourceModels()).isNull();
         assertThat(response.getResourceModel()).isEqualToComparingOnlyGivenFields(logGroup);
-        assertThat(response.getResourceModel().getTags()).isEqualTo(tags);
         assertThat(response.getMessage()).isNull();
         assertThat(response.getErrorCode()).isNull();
     }
@@ -236,38 +236,40 @@ public void handleRequest_SuccessNoChange_NoAction_WithPreviousModel() {
                 .retentionInDays(1)
                 .kmsKeyId("arn:aws:kms:us-east-1:$123456789012:key/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")
                 .build();
-        final Set<Tag> tags = new HashSet<>(Arrays.asList(
-                Tag.builder().key("key-1").value("value-1").build(),
-                Tag.builder().key("key-2").value("value-2").build()
-        ));
+        final Map<String, String> tags = new HashMap<String, String>() {{
+            put("key-1", "value-1");
+            put("key-2", "value-2");
+        }};
         final ResourceModel previousModel = ResourceModel.builder()
                 .logGroupName("LogGroup")
                 .retentionInDays(1)
                 .kmsKeyId("arn:aws:kms:us-east-1:$123456789012:key/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")
-                .tags(tags)
                 .build();
 
         final ResourceModel model = ResourceModel.builder()
                 .logGroupName("LogGroup")
                 .retentionInDays(1)
                 .kmsKeyId("arn:aws:kms:us-east-1:$123456789012:key/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")
-                .tags(tags)
                 .build();
 
         final ResourceHandlerRequest<ResourceModel> request = ResourceHandlerRequest.<ResourceModel>builder()
                 .previousResourceState(previousModel)
                 .desiredResourceState(model)
+                .previousResourceTags(tags)
+                .desiredResourceTags(tags)
                 .build();
 
         final ProgressEvent<ResourceModel, CallbackContext> response = handler.handleRequest(proxy, request, null, logger);
 
+        ArgumentCaptor<CloudWatchLogsRequest> requests = ArgumentCaptor.forClass(CloudWatchLogsRequest.class);
+        verify(proxy, times(0)).injectCredentialsAndInvokeV2(requests.capture(), any());
+
         assertThat(response).isNotNull();
         assertThat(response.getStatus()).isEqualTo(OperationStatus.SUCCESS);
         assertThat(response.getCallbackContext()).isNull();
         assertThat(response.getCallbackDelaySeconds()).isEqualTo(0);
         assertThat(response.getResourceModels()).isNull();
         assertThat(response.getResourceModel()).isEqualToComparingOnlyGivenFields(logGroup);
-        assertThat(response.getResourceModel().getTags()).isEqualTo(tags);
         assertThat(response.getMessage()).isNull();
         assertThat(response.getErrorCode()).isNull();
     }
@@ -279,25 +281,25 @@ public void handleRequest_Success_UpdateWith_RetentionAndKmsAndTags() {
             .retentionInDays(2)
             .kmsKeyId("arn:aws:kms:us-east-1:$123456789012:key/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")
             .build();
-        final Set<Tag> tags = new HashSet<>(Arrays.asList(
-                Tag.builder().key("key-1").value("value-1").build(),
-                Tag.builder().key("key-2").value("value-2").build()
-        ));
         final ResourceModel previousModel = ResourceModel.builder()
             .logGroupName("LogGroup")
             .build();
 
+        final Map<String, String> tags = new HashMap<String, String>() {{
+            put("key-1", "value-1");
+            put("key-2", "value-2");
+        }};
         final ResourceModel model = ResourceModel.builder()
                 .logGroupName("LogGroup")
                 .retentionInDays(2)
                 .kmsKeyId("arn:aws:kms:us-east-1:$123456789012:key/aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa")
-                .tags(tags)
                 .build();
 
         final ResourceHandlerRequest<ResourceModel> request = ResourceHandlerRequest.<ResourceModel>builder()
-            .previousResourceState(previousModel)
-            .desiredResourceState(model)
-            .build();
+                .previousResourceState(previousModel)
+                .desiredResourceState(model)
+                .desiredResourceTags(tags)
+                .build();
 
         final ProgressEvent<ResourceModel, CallbackContext> response = handler.handleRequest(proxy, request, null, logger);
 
@@ -315,7 +317,7 @@ public void handleRequest_Success_UpdateWith_RetentionAndKmsAndTags() {
 
         assertThat(requests.getAllValues().get(2)).isEqualTo(TagLogGroupRequest.builder()
                 .logGroupName("LogGroup")
-                .tags(Translator.translateTagsToSdk(tags))
+                .tags(tags)
                 .build());
 
         assertThat(response).isNotNull();
@@ -324,7 +326,6 @@ public void handleRequest_Success_UpdateWith_RetentionAndKmsAndTags() {
         assertThat(response.getCallbackDelaySeconds()).isEqualTo(0);
         assertThat(response.getResourceModels()).isNull();
         assertThat(response.getResourceModel()).isEqualToComparingOnlyGivenFields(logGroup);
-        assertThat(response.getResourceModel().getTags()).isEqualTo(tags);
         assertThat(response.getMessage()).isNull();
         assertThat(response.getErrorCode()).isNull();
     }
@@ -334,30 +335,29 @@ public void handleRequest_Success_AddTags() {
         final LogGroup logGroup = LogGroup.builder()
                 .logGroupName("LogGroup")
                 .build();
-        final Set<Tag> previousTags = new HashSet<>(Arrays.asList(
-                Tag.builder().key("key-1").value("value-1").build(),
-                Tag.builder().key("key-2").value("value-2").build()
-        ));
+        final Map<String, String> previousTags = new HashMap<String, String>() {{
+            put("key-1", "value-1");
+            put("key-2", "value-2");
+        }};
         final ResourceModel previousModel = ResourceModel.builder()
                 .logGroupName("LogGroup")
-                .tags(previousTags)
                 .build();
 
-        final Set<Tag> newTags = new HashSet<>(Arrays.asList(
-                Tag.builder().key("key-3").value("value-3").build(),
-                Tag.builder().key("key-4").value("value-4").build()
-        ));
-        final Set<Tag> tags = new HashSet<>();
-        tags.addAll(previousTags);
-        tags.addAll(newTags);
+        final Map<String, String> newTags = new HashMap<String, String>() {{
+            put("key-3", "value-3");
+            put("key-4", "value-4");
+        }};
+        final Map<String, String> tags = Stream.concat(previousTags.entrySet().stream(), newTags.entrySet().stream())
+                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
         final ResourceModel model = ResourceModel.builder()
                 .logGroupName("LogGroup")
-                .tags(tags)
                 .build();
 
         final ResourceHandlerRequest<ResourceModel> request = ResourceHandlerRequest.<ResourceModel>builder()
                 .previousResourceState(previousModel)
                 .desiredResourceState(model)
+                .previousResourceTags(previousTags)
+                .desiredResourceTags(tags)
                 .build();
 
         final ProgressEvent<ResourceModel, CallbackContext> response = handler.handleRequest(proxy, request, null, logger);
@@ -366,7 +366,7 @@ public void handleRequest_Success_AddTags() {
         verify(proxy, times(1)).injectCredentialsAndInvokeV2(requests.capture(), any());
         assertThat(requests.getAllValues().get(0)).isEqualTo(TagLogGroupRequest.builder()
                 .logGroupName("LogGroup")
-                .tags(Translator.translateTagsToSdk(newTags))
+                .tags(newTags)
                 .build());
 
         assertThat(response).isNotNull();
@@ -375,7 +375,6 @@ public void handleRequest_Success_AddTags() {
         assertThat(response.getCallbackDelaySeconds()).isEqualTo(0);
         assertThat(response.getResourceModels()).isNull();
         assertThat(response.getResourceModel()).isEqualToComparingOnlyGivenFields(logGroup);
-        assertThat(response.getResourceModel().getTags()).isEqualTo(tags);
         assertThat(response.getMessage()).isNull();
         assertThat(response.getErrorCode()).isNull();
     }
@@ -385,41 +384,41 @@ public void handleRequest_Success_UpdateTags() {
         final LogGroup logGroup = LogGroup.builder()
                 .logGroupName("LogGroup")
                 .build();
-        final Set<Tag> previousTags = new HashSet<>(Arrays.asList(
-                Tag.builder().key("key-1").value("value-1").build(),
-                Tag.builder().key("key-2").value("value-2").build()
-        ));
+        final Map<String, String> previousTags = new HashMap<String, String>() {{
+            put("key-1", "value-1");
+            put("key-2", "value-2");
+        }};
         final ResourceModel previousModel = ResourceModel.builder()
                 .logGroupName("LogGroup")
-                .tags(previousTags)
                 .build();
 
-        final Set<Tag> tags = new HashSet<>(Arrays.asList(
-                Tag.builder().key("key-2").value("value-2-new").build(),
-                Tag.builder().key("key-3").value("value-3").build()
-        ));
+        final Map<String, String> tags = new HashMap<String, String>() {{
+            put("key-2", "value-2-new");
+            put("key-3", "value-3");
+        }};
         final ResourceModel model = ResourceModel.builder()
                 .logGroupName("LogGroup")
-                .tags(tags)
                 .build();
 
         final ResourceHandlerRequest<ResourceModel> request = ResourceHandlerRequest.<ResourceModel>builder()
                 .previousResourceState(previousModel)
                 .desiredResourceState(model)
+                .previousResourceTags(previousTags)
+                .desiredResourceTags(tags)
                 .build();
 
         final ProgressEvent<ResourceModel, CallbackContext> response = handler.handleRequest(proxy, request, null, logger);
 
         ArgumentCaptor<CloudWatchLogsRequest> requests = ArgumentCaptor.forClass(CloudWatchLogsRequest.class);
         verify(proxy, times(2)).injectCredentialsAndInvokeV2(requests.capture(), any());
-        final List<String> removedTagKeys = Sets.difference(previousTags, tags).stream().map(Tag::getKey).collect(Collectors.toList());
+        final List<String> removedTagKeys = new ArrayList<>(Maps.difference(previousTags, tags).entriesOnlyOnLeft().keySet());
         assertThat(requests.getAllValues().get(0)).isEqualTo(UntagLogGroupRequest.builder()
                 .logGroupName("LogGroup")
                 .tags(removedTagKeys)
                 .build());
         assertThat(requests.getAllValues().get(1)).isEqualTo(TagLogGroupRequest.builder()
                 .logGroupName("LogGroup")
-                .tags(Translator.translateTagsToSdk(tags))
+                .tags(tags)
                 .build());
 
         assertThat(response).isNotNull();
@@ -428,7 +427,6 @@ public void handleRequest_Success_UpdateTags() {
         assertThat(response.getCallbackDelaySeconds()).isEqualTo(0);
         assertThat(response.getResourceModels()).isNull();
         assertThat(response.getResourceModel()).isEqualToComparingOnlyGivenFields(logGroup);
-        assertThat(response.getResourceModel().getTags()).isEqualTo(tags);
         assertThat(response.getMessage()).isNull();
         assertThat(response.getErrorCode()).isNull();
     }
@@ -438,13 +436,12 @@ public void handleRequest_Success_RemoveTags() {
         final LogGroup logGroup = LogGroup.builder()
                 .logGroupName("LogGroup")
                 .build();
-        final Set<Tag> tags = new HashSet<>(Arrays.asList(
-                Tag.builder().key("key-1").value("value-1").build(),
-                Tag.builder().key("key-2").value("value-2").build()
-        ));
+        final Map<String, String> previousTags = new HashMap<String, String>() {{
+            put("key-1", "value-1");
+            put("key-2", "value-2");
+        }};
         final ResourceModel previousModel = ResourceModel.builder()
                 .logGroupName("LogGroup")
-                .tags(tags)
                 .build();
 
         final ResourceModel model = ResourceModel.builder()
@@ -454,13 +451,14 @@ public void handleRequest_Success_RemoveTags() {
         final ResourceHandlerRequest<ResourceModel> request = ResourceHandlerRequest.<ResourceModel>builder()
                 .previousResourceState(previousModel)
                 .desiredResourceState(model)
+                .previousResourceTags(previousTags)
                 .build();
 
         final ProgressEvent<ResourceModel, CallbackContext> response = handler.handleRequest(proxy, request, null, logger);
 
         ArgumentCaptor<CloudWatchLogsRequest> requests = ArgumentCaptor.forClass(CloudWatchLogsRequest.class);
         verify(proxy, times(1)).injectCredentialsAndInvokeV2(requests.capture(), any());
-        final List<String> removedTagKeys = tags.stream().map(Tag::getKey).collect(Collectors.toList());
+        final List<String> removedTagKeys = new ArrayList<>(previousTags.keySet());
         assertThat(requests.getAllValues().get(0)).isEqualTo(UntagLogGroupRequest.builder()
                 .logGroupName("LogGroup")
                 .tags(removedTagKeys)
@@ -472,7 +470,6 @@ public void handleRequest_Success_RemoveTags() {
         assertThat(response.getCallbackDelaySeconds()).isEqualTo(0);
         assertThat(response.getResourceModels()).isNull();
         assertThat(response.getResourceModel()).isEqualToComparingOnlyGivenFields(logGroup);
-        assertThat(response.getResourceModel().getTags()).isNull();
         assertThat(response.getMessage()).isNull();
         assertThat(response.getErrorCode()).isNull();
     }
@@ -792,18 +789,18 @@ public void handleRequest_AddTags_FailureNotFound_ServiceException() {
                         any()
                 );
 
-        final Set<Tag> tags = new HashSet<>(Arrays.asList(
-                Tag.builder().key("key-1").value("value-1").build(),
-                Tag.builder().key("key-2").value("value-2").build()
-        ));
+        final Map<String, String> tags = new HashMap<String, String>() {{
+            put("key-1", "value-1");
+            put("key-2", "value-2");
+        }};
         final ResourceModel model = ResourceModel.builder()
                 .logGroupName("LogGroup")
-                .tags(tags)
                 .build();
 
         final ResourceHandlerRequest<ResourceModel> request = ResourceHandlerRequest.<ResourceModel>builder()
                 .previousResourceState(previousModel)
                 .desiredResourceState(model)
+                .desiredResourceTags(tags)
                 .build();
 
         assertThrows(software.amazon.cloudformation.exceptions.ResourceNotFoundException.class,
@@ -822,15 +819,15 @@ public void handleRequest_AddTags_InvalidParameter_ServiceException() {
                         any()
                 );
 
-        final Set<Tag> tags = Collections.singleton(Tag.builder().key("key-1").value("value-1").build());
+        final Map<String, String> tags = Collections.singletonMap("key-1", "value-1");
         final ResourceModel model = ResourceModel.builder()
                 .logGroupName("LogGroup")
-                .tags(tags)
                 .build();
 
         final ResourceHandlerRequest<ResourceModel> request = ResourceHandlerRequest.<ResourceModel>builder()
                 .previousResourceState(previousModel)
                 .desiredResourceState(model)
+                .desiredResourceTags(tags)
                 .build();
 
         assertThrows(software.amazon.cloudformation.exceptions.CfnInternalFailureException.class,
@@ -839,13 +836,12 @@ public void handleRequest_AddTags_InvalidParameter_ServiceException() {
 
     @Test
     public void handleRequest_RemoveTags_FailureNotFound_ServiceException() {
-        final Set<Tag> previousTags = new HashSet<>(Arrays.asList(
-                Tag.builder().key("key-1").value("value-1").build(),
-                Tag.builder().key("key-2").value("value-2").build()
-        ));
+        final Map<String, String> previousTags = new HashMap<String, String>() {{
+            put("key-1", "value-1");
+            put("key-2", "value-2");
+        }};
         final ResourceModel previousModel = ResourceModel.builder()
                 .logGroupName("LogGroup")
-                .tags(previousTags)
                 .build();
         doThrow(software.amazon.awssdk.services.cloudwatchlogs.model.ResourceNotFoundException.class)
                 .when(proxy)
@@ -861,6 +857,7 @@ public void handleRequest_RemoveTags_FailureNotFound_ServiceException() {
         final ResourceHandlerRequest<ResourceModel> request = ResourceHandlerRequest.<ResourceModel>builder()
                 .previousResourceState(previousModel)
                 .desiredResourceState(model)
+                .previousResourceTags(previousTags)
                 .build();
 
         assertThrows(software.amazon.cloudformation.exceptions.ResourceNotFoundException.class,
@@ -870,10 +867,9 @@ public void handleRequest_RemoveTags_FailureNotFound_ServiceException() {
 
     @Test
     public void handleRequest_RemoveTags_InvalidParameter_ServiceException() {
-        final Set<Tag> tags = Collections.singleton(Tag.builder().key("key-1").value("value-1").build());
+        final Map<String, String> previousTags = Collections.singletonMap("key-1", "value-1");
         final ResourceModel previousModel = ResourceModel.builder()
                 .logGroupName("LogGroup")
-                .tags(tags)
                 .build();
         doThrow(software.amazon.awssdk.services.cloudwatchlogs.model.InvalidParameterException.class)
                 .when(proxy)
@@ -889,6 +885,7 @@ public void handleRequest_RemoveTags_InvalidParameter_ServiceException() {
         final ResourceHandlerRequest<ResourceModel> request = ResourceHandlerRequest.<ResourceModel>builder()
                 .previousResourceState(previousModel)
                 .desiredResourceState(model)
+                .previousResourceTags(previousTags)
                 .build();
 
         assertThrows(software.amazon.cloudformation.exceptions.CfnInternalFailureException.class,