Fix BatchGetItem & TransactGetItems operations failing on the DynamoDB plugin (#3120)

* Fix batch operations APIs failing with DynamoDB

* Fixed TransactGetItems operation on DynamoDB

app/server/appsmith-plugins/dynamoPlugin/src/main/java/com/external/plugins/DynamoPlugin.java

@@ -36,6 +36,8 @@ import java.io.IOException;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.lang.reflect.WildcardType;
 import java.net.URI;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -115,7 +117,8 @@ public class DynamoPlugin extends BasePlugin {
                             toLowerCamelCase(action),
                             requestClass
                     );
-                    final DynamoDbResponse response = (DynamoDbResponse) actionExecuteMethod.invoke(ddb, plainToSdk(parameters, requestClass));
+                    final Object sdkValue = plainToSdk(parameters, requestClass);
+                    final DynamoDbResponse response = (DynamoDbResponse) actionExecuteMethod.invoke(ddb, sdkValue);
                     result.setBody(sdkToPlain(response));
                 } catch (AppsmithPluginException | InvocationTargetException | IllegalAccessException | NoSuchMethodException | ClassNotFoundException e) {
                     final String message = "Error executing the DynamoDB Action: " + (e.getCause() == null ? e : e.getCause()).getMessage();
@@ -290,34 +293,43 @@ public class DynamoPlugin extends BasePlugin {
                     // For maps, we go recursive, applying this transformation to each value, and replacing with the
                     // result in the map. Generic types in the setter method's signature are used to convert the values.
                     final Method setterMethod = findMethod(builderType, m -> m.getName().equals(setterName));
-                    final ParameterizedType valueType = (ParameterizedType) setterMethod.getGenericParameterTypes()[0];
+                    final Type parameterType = setterMethod.getGenericParameterTypes()[0];
-                    final Map<String, Object> transformedMap = new HashMap<>();
+                    if (parameterType instanceof ParameterizedType) {
-                    for (final Map.Entry<String, Object> innerEntry : ((Map<String, Object>) value).entrySet()) {
+                        final ParameterizedType valueType = (ParameterizedType) parameterType;
-                        Object innerValue = innerEntry.getValue();
+                        final Map<String, Object> transformedMap = new HashMap<>();
-                        if (innerValue instanceof Map) {
+                        for (final Map.Entry<String, Object> innerEntry : ((Map<String, Object>) value).entrySet()) {
-                            innerValue = plainToSdk((Map) innerValue, (Class<?>) valueType.getActualTypeArguments()[1]);
+                            Object innerValue = innerEntry.getValue();
+                            if (innerValue instanceof Map) {
+                                innerValue = plainToSdk((Map) innerValue, (Class<?>) valueType.getActualTypeArguments()[1]);
+                            }
+                            transformedMap.put(innerEntry.getKey(), innerValue);
                         }
-                        transformedMap.put(innerEntry.getKey(), innerValue);
+                        value = transformedMap;
+                        if (!Map.class.isAssignableFrom((Class<?>) valueType.getRawType())) {
+                            // Some setters don't take a plain map. For example, some require an `AttributeValue` instance
+                            // for objects that are just maps in JSON. So, we make that conversion here.
+                            value = plainToSdk((Map) value, (Class<T>) valueType.getRawType());
+                        }
+                        setterMethod.invoke(builder, value);
+                    } else if (parameterType instanceof Class) {
+                        setterMethod.invoke(builder, plainToSdk((Map) value, (Class) parameterType));
                     }
-                    value = transformedMap;
-                    if (!Map.class.isAssignableFrom((Class<?>) valueType.getRawType())) {
-                        // Some setters don't take a plain map. For example, some require an `AttributeValue` instance
-                        // for objects that are just maps in JSON. So, we make that conversion here.
-                        value = plainToSdk((Map) value, (Class<T>) valueType.getRawType());
-                    }
-                    setterMethod.invoke(builder, value);
 
                 } else if (value instanceof Collection) {
                     // For linear collections, the process is similar to that of maps.
                     final Collection<Object> valueAsCollection = (Collection) value;
                     // Find method by name and exclude the varargs version of the method.
                     final Method setterMethod = findMethod(builderType, m -> m.getName().equals(setterName) && !m.getParameterTypes()[0].getName().startsWith("[L"));
-                    final ParameterizedType valueType = (ParameterizedType) setterMethod.getGenericParameterTypes()[0];
+                    Type valueType = ((ParameterizedType) setterMethod.getGenericParameterTypes()[0]).getActualTypeArguments()[0];
+                    if (valueType instanceof WildcardType) {
+                        // This occurs when the method's parameter is typed as `Collection<? extends Map<...>>`. Example op: `BatchGetItem`.
+                        valueType = ((WildcardType) valueType).getUpperBounds()[0];
+                    }
                     final Collection<Object> reTypedList = new ArrayList<>();
                     for (final Object innerValue : valueAsCollection) {
                         if (innerValue instanceof Map) {
-                            reTypedList.add(plainToSdk((Map) innerValue, (Class<?>) valueType.getActualTypeArguments()[0]));
+                            reTypedList.add(plainToSdk((Map) innerValue, valueType));
-                        } else if (innerValue instanceof String && SdkBytes.class.isAssignableFrom((Class<?>) valueType.getActualTypeArguments()[0])) {
+                        } else if (innerValue instanceof String && SdkBytes.class.isAssignableFrom((Class<?>) valueType)) {
                             reTypedList.add(SdkBytes.fromUtf8String((String) innerValue));
                         } else {
                             reTypedList.add(innerValue);
@@ -338,6 +350,34 @@ public class DynamoPlugin extends BasePlugin {
         return (T) builderType.getMethod("build").invoke(builder);
     }
 
+    public static Object plainToSdk(Map<String, Object> mapping, Type type)
+            throws InvocationTargetException, NoSuchMethodException, ClassNotFoundException, AppsmithPluginException,
+            IllegalAccessException {
+
+        if (mapping == null) {
+            return null;
+        }
+
+        if (!(type instanceof ParameterizedType)) {
+            return plainToSdk(mapping, (Class) type);
+        }
+
+        final ParameterizedType ptype = (ParameterizedType) type;
+
+        if (Map.class.equals(ptype.getRawType())) {
+            final Map<String, Object> convertedMap = new HashMap<>();
+            for (final Map.Entry<String, Object> entry : mapping.entrySet()) {
+                convertedMap.put(entry.getKey(), plainToSdk((Map) entry.getValue(), (Class<?>) ptype.getActualTypeArguments()[1]));
+            }
+            return convertedMap;
+        }
+
+        throw new AppsmithPluginException(
+                AppsmithPluginError.PLUGIN_ERROR,
+                "Unknown type to convert to SDK style " + type.getTypeName()
+        );
+    }
+
     private static Method findMethod(Class<?> builderType, Predicate<Method> predicate) {
         return Arrays.stream(builderType.getMethods())
                 .filter(predicate)

app/server/appsmith-plugins/dynamoPlugin/src/test/java/com/external/plugins/DynamoPluginTest.java

@@ -27,6 +27,7 @@ import software.amazon.awssdk.services.dynamodb.model.PutItemRequest;
 import software.amazon.awssdk.services.dynamodb.model.ScalarAttributeType;
 
 import java.net.URI;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
@@ -238,6 +239,74 @@ public class DynamoPluginTest {
                 .verifyComplete();
     }
 
+    @Test
+    public void testBatchGetItem() {
+        final String body = "{\n" +
+                "    \"RequestItems\": {\n" +
+                "        \"cities\": {\n" +
+                "            \"Keys\": [\n" +
+                "                {\n" +
+                "                    \"Id\": {\n" +
+                "                       \"S\": \"1\"\n" +
+                "                    }\n" +
+                "                },\n" +
+                "                {\n" +
+                "                    \"Id\": {\n" +
+                "                       \"S\": \"2\"\n" +
+                "                    }\n" +
+                "                }\n" +
+                "            ],\n" +
+                "            \"ProjectionExpression\":\"City\"\n" +
+                "        }\n" +
+                "    },\n" +
+                "    \"ReturnConsumedCapacity\": \"TOTAL\"\n" +
+                "}";
+
+        StepVerifier.create(execute("BatchGetItem", body))
+                .assertNext(result -> {
+                    assertNotNull(result);
+                    assertTrue(result.getIsExecutionSuccess());
+                    final Map<String, ?> response = (Map) result.getBody();
+                    assertEquals(
+                            Collections.emptyMap(),
+                            response.remove("UnprocessedKeys")
+                    );
+                    final List<Map<String, Map<String, String>>> items = (List<Map<String, Map<String, String>>>) ((Map) response.get("Responses")).get("cities");
+                    assertEquals("New Delhi", items.get(0).get("City").get("S"));
+                    assertEquals("Bengaluru", items.get(1).get("City").get("S"));
+                })
+                .verifyComplete();
+    }
+
+    @Test
+    public void testTransactGetItems() {
+        final String body =
+                "{\n" +
+                "  \"ReturnConsumedCapacity\": \"NONE\",\n" +
+                "  \"TransactItems\": [\n" +
+                "    {\n" +
+                "      \"Get\": {\n" +
+                "        \"Key\": {\n" +
+                "          \"Id\": {\n" +
+                "            \"S\": \"1\"\n" +
+                "          }\n" +
+                "        },\n" +
+                "        \"TableName\": \"cities\"\n" +
+                "      }\n" +
+                "    }\n" +
+                "  ]\n" +
+                "}";
+
+        StepVerifier.create(execute("TransactGetItems", body))
+                .assertNext(result -> {
+                    assertNotNull(result);
+                    assertTrue(result.getIsExecutionSuccess());
+                    final Map<String, ?> response = (Map) result.getBody();
+                    assertEquals("New Delhi", ((List<Map<String, Map<String, Map<String, String>>>>) response.get("Responses")).get(0).get("Item").get("City").get("S"));
+                })
+                .verifyComplete();
+    }
+
     @Test
     public void testStructure() {
         final Mono<DatasourceStructure> structureMono = pluginExecutor