Repository: getmoto/moto
Pull request #5840 (2023-01-13 07:10:48)

batch: add multinode support
This adds multinode support to `mock_batch`. This will allow users to run jobs that have multiple different jobs configured.

Notes:
* describe_jobs doesn't return most container field
* describe_jobs doesn't respect the `#idx` suffix to job IDs
* all containers log to the same stream
* container exit logic is very simple and waits for them to finish in order so if replica 1 fails but 0 doesn't it may hang until timeout
* there's minimal/no validation on the job spec right now for nodeProperties

Test plan:

pyre

added multinode slow test

kiukchung: this is pretty amazing! lets us test distributed AI workloads (that are submitted as batch's multi-node-parallel jobs)

Patch:
diff --git a/moto/apigateway/models.py b/moto/apigateway/models.py
index d4497e0c4..e0fa232d5 100644
--- a/moto/apigateway/models.py
+++ b/moto/apigateway/models.py
@@ -5,6 +5,7 @@ import re
 import responses
 import requests
 import time
+from datetime import datetime
 from collections import defaultdict
 from openapi_spec_validator import validate_spec
 from typing import Any, Dict, List, Optional, Tuple, Union
@@ -19,6 +20,7 @@ from moto.core import BaseBackend, BackendDict, BaseModel, CloudFormationModel
 from .utils import create_id, to_path
 from moto.core.utils import path_url
 from .exceptions import (
+    BadRequestException,
     ConflictException,
     DeploymentNotFoundException,
     ApiKeyNotFoundException,
@@ -1572,6 +1574,42 @@ class APIGatewayBackend(BaseBackend):
         self.put_rest_api(api.id, api_doc, fail_on_warnings=fail_on_warnings)
         return api
 
+    def export_api(self, rest_api_id: str, export_type: str) -> Dict[str, Any]:
+        """
+        Not all fields are implemented yet.
+        The export-type is currently ignored - we will only return the 'swagger'-format
+        """
+        try:
+            api = self.get_rest_api(rest_api_id)
+        except RestAPINotFound:
+            raise StageNotFoundException
+        if export_type not in ["swagger", "oas30"]:
+            raise BadRequestException(f"No API exporter for type '{export_type}'")
+        now = datetime.now().strftime("%Y-%m-%dT%H:%m:%S")
+        resp: Dict[str, Any] = {
+            "swagger": "2.0",
+            "info": {"version": now, "title": api.name},
+            "host": f"{api.id}.execute-api.{self.region_name}.amazonaws.com",
+            "basePath": "/",
+            "schemes": ["https"],
+            "paths": {},
+            "definitions": {"Empty": {"type": "object", "title": "Empty Schema"}},
+        }
+        for res in api.resources.values():
+            path = res.get_path()
+            resp["paths"][path] = {}
+            for method_type, method in res.resource_methods.items():
+                resp["paths"][path][method_type] = {
+                    "produces": ["application/json"],
+                    "responses": {},
+                }
+                for code, _ in method.method_responses.items():
+                    resp["paths"][path][method_type]["responses"][code] = {
+                        "description": f"{code} response",
+                        "schema": {"$ref": "#/definitions/Empty"},
+                    }
+        return resp
+
     def get_rest_api(self, function_id: str) -> RestAPI:
         rest_api = self.apis.get(function_id)
         if rest_api is None:
@@ -1610,6 +1648,27 @@ class APIGatewayBackend(BaseBackend):
         for (path, resource_doc) in sorted(
             api_doc["paths"].items(), key=lambda x: x[0]
         ):
+            # We may want to create a path like /store/inventory
+            # Ensure that /store exists first, so we can use it as a parent
+            ancestors = path.split("/")[
+                1:-1
+            ]  # skip first (empty), skip last (child) - only process ancestors
+            direct_parent = ""
+            parent_id = self.apis[function_id].get_resource_for_path("/").id
+            for a in ancestors:
+                res = self.apis[function_id].get_resource_for_path(
+                    direct_parent + "/" + a
+                )
+                if res is None:
+                    res = self.create_resource(
+                        function_id=function_id,
+                        parent_resource_id=parent_id,
+                        path_part=a,
+                    )
+                parent_id = res.id
+                direct_parent = direct_parent + "/" + a
+
+            # Now that we know all ancestors are created, create the resource itself
             parent_path_part = path[0 : path.rfind("/")] or "/"
             parent_resource_id = (
                 self.apis[function_id].get_resource_for_path(parent_path_part).id
diff --git a/moto/apigateway/responses.py b/moto/apigateway/responses.py
index 2e22c5854..13b96d4f9 100644
--- a/moto/apigateway/responses.py
+++ b/moto/apigateway/responses.py
@@ -429,6 +429,22 @@ class APIGatewayResponse(BaseResponse):
             self.backend.delete_stage(function_id, stage_name)
             return 202, {}, "{}"
 
+    def export(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE:  # type: ignore[return]
+        self.setup_class(request, full_url, headers)
+        url_path_parts = self.path.split("/")
+        rest_api_id = url_path_parts[-5]
+        export_type = url_path_parts[-1]
+
+        body = self.backend.export_api(rest_api_id, export_type)
+
+        now = body["info"]["version"]
+        filename = f"swagger_{now}Z.json"
+        headers = {
+            "Content-Type": "application/octet-stream",
+            "Content-Disposition": f'attachment; filename="{filename}"',
+        }
+        return 200, headers, json.dumps(body).encode("utf-8")
+
     def integrations(self, request: Any, full_url: str, headers: Dict[str, str]) -> TYPE_RESPONSE:  # type: ignore[return]
         self.setup_class(request, full_url, headers)
         url_path_parts = self.path.split("/")
diff --git a/moto/apigateway/urls.py b/moto/apigateway/urls.py
index 461d899bb..d2961c6f6 100644
--- a/moto/apigateway/urls.py
+++ b/moto/apigateway/urls.py
@@ -14,6 +14,7 @@ url_paths = {
     "{0}/restapis/(?P<function_id>[^/]+)/stages$": response.restapis_stages,
     "{0}/tags/arn:aws:apigateway:(?P<region_name>[^/]+)::/restapis/(?P<function_id>[^/]+)/stages/(?P<stage_name>[^/]+)/?$": response.restapis_stages_tags,
     "{0}/restapis/(?P<function_id>[^/]+)/stages/(?P<stage_name>[^/]+)/?$": response.stages,
+    "{0}/restapis/(?P<function_id>[^/]+)/stages/(?P<stage_name>[^/]+)/exports/(?P<export_type>[^/]+)/?$": response.export,
     "{0}/restapis/(?P<function_id>[^/]+)/deployments$": response.deployments,
     "{0}/restapis/(?P<function_id>[^/]+)/deployments/(?P<deployment_id>[^/]+)/?$": response.individual_deployment,
     "{0}/restapis/(?P<function_id>[^/]+)/resources/(?P<resource_id>[^/]+)/?$": response.resource_individual,
diff --git a/moto/batch/models.py b/moto/batch/models.py
index 01c4a71fd..09d3fd3c9 100644
--- a/moto/batch/models.py
+++ b/moto/batch/models.py
@@ -21,6 +21,7 @@ from .exceptions import InvalidParameterValueException, ClientException, Validat
 from .utils import (
     make_arn_for_compute_env,
     make_arn_for_job_queue,
+    make_arn_for_job,
     make_arn_for_task_def,
     lowercase_first_key,
 )
@@ -221,6 +222,7 @@ class JobDefinition(CloudFormationModel):
         parameters: Optional[Dict[str, Any]],
         _type: str,
         container_properties: Dict[str, Any],
+        node_properties: Dict[str, Any],
         tags: Dict[str, str],
         retry_strategy: Dict[str, str],
         timeout: Dict[str, int],
@@ -235,6 +237,7 @@ class JobDefinition(CloudFormationModel):
         self.revision = revision or 0
         self._region = backend.region_name
         self.container_properties = container_properties
+        self.node_properties = node_properties
         self.status = "ACTIVE"
         self.parameters = parameters or {}
         self.timeout = timeout
@@ -242,10 +245,11 @@ class JobDefinition(CloudFormationModel):
         self.platform_capabilities = platform_capabilities
         self.propagate_tags = propagate_tags
 
-        if "resourceRequirements" not in self.container_properties:
-            self.container_properties["resourceRequirements"] = []
-        if "secrets" not in self.container_properties:
-            self.container_properties["secrets"] = []
+        if self.container_properties is not None:
+            if "resourceRequirements" not in self.container_properties:
+                self.container_properties["resourceRequirements"] = []
+            if "secrets" not in self.container_properties:
+                self.container_properties["secrets"] = []
 
         self._validate()
         self.revision += 1
@@ -306,26 +310,28 @@ class JobDefinition(CloudFormationModel):
 
     def _validate(self) -> None:
         # For future use when containers arnt the only thing in batch
-        if self.type not in ("container",):
-            raise ClientException('type must be one of "container"')
+        VALID_TYPES = ("container", "multinode")
+        if self.type not in VALID_TYPES:
+            raise ClientException(f"type must be one of {VALID_TYPES}")
 
         if not isinstance(self.parameters, dict):
             raise ClientException("parameters must be a string to string map")
 
-        if "image" not in self.container_properties:
-            raise ClientException("containerProperties must contain image")
+        if self.type == "container":
+            if "image" not in self.container_properties:
+                raise ClientException("containerProperties must contain image")
 
-        memory = self._get_resource_requirement("memory")
-        if memory is None:
-            raise ClientException("containerProperties must contain memory")
-        if memory < 4:
-            raise ClientException("container memory limit must be greater than 4")
+            memory = self._get_resource_requirement("memory")
+            if memory is None:
+                raise ClientException("containerProperties must contain memory")
+            if memory < 4:
+                raise ClientException("container memory limit must be greater than 4")
 
-        vcpus = self._get_resource_requirement("vcpus")
-        if vcpus is None:
-            raise ClientException("containerProperties must contain vcpus")
-        if vcpus <= 0:
-            raise ClientException("container vcpus limit must be greater than 0")
+            vcpus = self._get_resource_requirement("vcpus")
+            if vcpus is None:
+                raise ClientException("containerProperties must contain vcpus")
+            if vcpus <= 0:
+                raise ClientException("container vcpus limit must be greater than 0")
 
     def deregister(self) -> None:
         self.status = "INACTIVE"
@@ -335,6 +341,7 @@ class JobDefinition(CloudFormationModel):
         parameters: Optional[Dict[str, Any]],
         _type: str,
         container_properties: Dict[str, Any],
+        node_properties: Dict[str, Any],
         retry_strategy: Dict[str, Any],
         tags: Dict[str, str],
         timeout: Dict[str, int],
@@ -357,6 +364,7 @@ class JobDefinition(CloudFormationModel):
             parameters,
             _type,
             container_properties,
+            node_properties=node_properties,
             revision=self.revision,
             retry_strategy=retry_strategy,
             tags=tags,
@@ -416,7 +424,16 @@ class JobDefinition(CloudFormationModel):
             _type="container",
             tags=lowercase_first_key(properties.get("Tags", {})),
             retry_strategy=lowercase_first_key(properties["RetryStrategy"]),
-            container_properties=lowercase_first_key(properties["ContainerProperties"]),
+            container_properties=(
+                lowercase_first_key(properties["ContainerProperties"])
+                if "ContainerProperties" in properties
+                else None
+            ),
+            node_properties=(
+                lowercase_first_key(properties["NodeProperties"])
+                if "NodeProperties" in properties
+                else None
+            ),
             timeout=lowercase_first_key(properties.get("timeout", {})),
             platform_capabilities=None,
             propagate_tags=None,
@@ -466,6 +483,10 @@ class Job(threading.Thread, BaseModel, DockerModel, ManagedState):
         self.timeout = timeout
         self.all_jobs = all_jobs
 
+        self.arn = make_arn_for_job(
+            job_def.backend.account_id, self.job_id, job_def._region
+        )
+
         self.stop = False
         self.exit_code: Optional[int] = None
 
@@ -483,6 +504,7 @@ class Job(threading.Thread, BaseModel, DockerModel, ManagedState):
     def describe_short(self) -> Dict[str, Any]:
         result = {
             "jobId": self.job_id,
+            "jobArn": self.arn,
             "jobName": self.job_name,
             "createdAt": datetime2int_milliseconds(self.job_created_at),
             "status": self.status,
@@ -502,7 +524,13 @@ class Job(threading.Thread, BaseModel, DockerModel, ManagedState):
         result = self.describe_short()
         result["jobQueue"] = self.job_queue.arn
         result["dependsOn"] = self.depends_on or []
-        result["container"] = self._container_details()
+        if self.job_definition.type == "container":
+            result["container"] = self._container_details()
+        elif self.job_definition.type == "multinode":
+            result["container"] = {
+                "logStreamName": self.log_stream_name,
+            }
+            result["nodeProperties"] = self.job_definition.node_properties
         if self.job_stopped:
             result["stoppedAt"] = datetime2int_milliseconds(self.job_stopped_at)
         if self.timeout:
@@ -575,6 +603,8 @@ class Job(threading.Thread, BaseModel, DockerModel, ManagedState):
         try:
             import docker
 
+            containers: List[docker.models.containers.Container] = []
+
             self.advance()
             while self.status == "SUBMITTED":
                 # Wait until we've moved onto state 'PENDING'
@@ -585,34 +615,87 @@ class Job(threading.Thread, BaseModel, DockerModel, ManagedState):
             if self.depends_on and not self._wait_for_dependencies():
                 return
 
-            image = self.job_definition.container_properties.get(
-                "image", "alpine:latest"
-            )
-            privileged = self.job_definition.container_properties.get(
-                "privileged", False
-            )
-            cmd = self._get_container_property(
-                "command",
-                '/bin/sh -c "for a in `seq 1 10`; do echo Hello World; sleep 1; done"',
-            )
-            environment = {
-                e["name"]: e["value"]
-                for e in self._get_container_property("environment", [])
-            }
-            volumes = {
-                v["name"]: v["host"]
-                for v in self._get_container_property("volumes", [])
-            }
-            mounts = [
-                docker.types.Mount(
-                    m["containerPath"],
-                    volumes[m["sourceVolume"]]["sourcePath"],
-                    type="bind",
-                    read_only=m["readOnly"],
+            container_kwargs = []
+            if self.job_definition.container_properties:
+                volumes = {
+                    v["name"]: v["host"]
+                    for v in self._get_container_property("volumes", [])
+                }
+                container_kwargs.append(
+                    {
+                        "image": self.job_definition.container_properties.get(
+                            "image", "alpine:latest"
+                        ),
+                        "privileged": self.job_definition.container_properties.get(
+                            "privileged", False
+                        ),
+                        "command": self._get_container_property(
+                            "command",
+                            '/bin/sh -c "for a in `seq 1 10`; do echo Hello World; sleep 1; done"',
+                        ),
+                        "environment": {
+                            e["name"]: e["value"]
+                            for e in self._get_container_property("environment", [])
+                        },
+                        "mounts": [
+                            docker.types.Mount(
+                                m["containerPath"],
+                                volumes[m["sourceVolume"]]["sourcePath"],
+                                type="bind",
+                                read_only=m["readOnly"],
+                            )
+                            for m in self._get_container_property("mountPoints", [])
+                        ],
+                        "name": f"{self.job_name}-{self.job_id}",
+                    }
                 )
-                for m in self._get_container_property("mountPoints", [])
-            ]
-            name = f"{self.job_name}-{self.job_id}"
+            else:
+                node_properties = self.job_definition.node_properties
+                num_nodes = node_properties["numNodes"]
+                node_containers = {}
+                for node_range in node_properties["nodeRangeProperties"]:
+                    start, sep, end = node_range["targetNodes"].partition(":")
+                    if sep == "":
+                        start = end = int(start)
+                    else:
+                        if start == "":
+                            start = 0
+                        else:
+                            start = int(start)
+                        if end == "":
+                            end = num_nodes - 1
+                        else:
+                            end = int(end)
+                    for i in range(start, end + 1):
+                        node_containers[i] = node_range["container"]
+
+                for i in range(num_nodes):
+                    spec = node_containers[i]
+                    volumes = {v["name"]: v["host"] for v in spec.get("volumes", [])}
+                    container_kwargs.append(
+                        {
+                            "image": spec.get("image", "alpine:latest"),
+                            "privileged": spec.get("privileged", False),
+                            "command": spec.get(
+                                "command",
+                                '/bin/sh -c "for a in `seq 1 10`; do echo Hello World; sleep 1; done"',
+                            ),
+                            "environment": {
+                                e["name"]: e["value"]
+                                for e in spec.get("environment", [])
+                            },
+                            "mounts": [
+                                docker.types.Mount(
+                                    m["containerPath"],
+                                    volumes[m["sourceVolume"]]["sourcePath"],
+                                    type="bind",
+                                    read_only=m["readOnly"],
+                                )
+                                for m in spec.get("mountPoints", [])
+                            ],
+                            "name": f"{self.job_name}-{self.job_id}-{i}",
+                        }
+                    )
 
             self.advance()
             while self.status == "PENDING":
@@ -632,19 +715,20 @@ class Job(threading.Thread, BaseModel, DockerModel, ManagedState):
                 else {}
             )
 
-            environment["MOTO_HOST"] = settings.moto_server_host()
-            environment["MOTO_PORT"] = settings.moto_server_port()
-            environment[
-                "MOTO_HTTP_ENDPOINT"
-            ] = f'{environment["MOTO_HOST"]}:{environment["MOTO_PORT"]}'
-
-            run_kwargs = dict()
-            network_name = settings.moto_network_name()
-            network_mode = settings.moto_network_mode()
-            if network_name:
-                run_kwargs["network"] = network_name
-            elif network_mode:
-                run_kwargs["network_mode"] = network_mode
+            for kwargs in container_kwargs:
+                environment = kwargs["environment"]
+                environment["MOTO_HOST"] = settings.moto_server_host()
+                environment["MOTO_PORT"] = settings.moto_server_port()
+                environment[
+                    "MOTO_HTTP_ENDPOINT"
+                ] = f'{environment["MOTO_HOST"]}:{environment["MOTO_PORT"]}'
+
+                network_name = settings.moto_network_name()
+                network_mode = settings.moto_network_mode()
+                if network_name:
+                    kwargs["network"] = network_name
+                elif network_mode:
+                    kwargs["network_mode"] = network_mode
 
             log_config = docker.types.LogConfig(type=docker.types.LogConfig.types.JSON)
             self.advance()
@@ -656,107 +740,122 @@ class Job(threading.Thread, BaseModel, DockerModel, ManagedState):
             while self.status == "STARTING":
                 # Wait until the state is no longer runnable, but 'RUNNING'
                 sleep(0.5)
-            container = self.docker_client.containers.run(
-                image,
-                cmd,
-                detach=True,
-                name=name,
-                log_config=log_config,
-                environment=environment,
-                mounts=mounts,
-                privileged=privileged,
-                extra_hosts=extra_hosts,
-                **run_kwargs,
-            )
-            try:
-                container.reload()
 
-                max_time = None
-                if self._get_attempt_duration():
-                    attempt_duration = self._get_attempt_duration()
-                    max_time = self.job_started_at + datetime.timedelta(
-                        seconds=attempt_duration  # type: ignore[arg-type]
-                    )
+            for kwargs in container_kwargs:
+                if len(containers) > 0:
+                    env = kwargs["environment"]
+                    ip = containers[0].attrs["NetworkSettings"]["IPAddress"]
+                    env["AWS_BATCH_JOB_MAIN_NODE_PRIVATE_IPV4_ADDRESS"] = ip
+                container = self.docker_client.containers.run(
+                    detach=True,
+                    log_config=log_config,
+                    extra_hosts=extra_hosts,
+                    **kwargs,
+                )
+                container.reload()
+                containers.append(container)
 
-                while container.status == "running" and not self.stop:
+            for i, container in enumerate(containers):
+                try:
                     container.reload()
-                    time.sleep(0.5)
 
-                    if max_time and datetime.datetime.now() > max_time:
-                        raise Exception(
-                            "Job time exceeded the configured attemptDurationSeconds"
+                    max_time = None
+                    if self._get_attempt_duration():
+                        attempt_duration = self._get_attempt_duration()
+                        max_time = self.job_started_at + datetime.timedelta(
+                            seconds=attempt_duration  # type: ignore[arg-type]
                         )
 
-                # Container should be stopped by this point... unless asked to stop
-                if container.status == "running":
-                    container.kill()
-
-                # Log collection
-                logs_stdout = []
-                logs_stderr = []
-                logs_stderr.extend(
-                    container.logs(
-                        stdout=False,
-                        stderr=True,
-                        timestamps=True,
-                        since=datetime2int(self.job_started_at),
+                    while container.status == "running" and not self.stop:
+                        container.reload()
+                        time.sleep(0.5)
+
+                        if max_time and datetime.datetime.now() > max_time:
+                            raise Exception(
+                                "Job time exceeded the configured attemptDurationSeconds"
+                            )
+
+                    # Container should be stopped by this point... unless asked to stop
+                    if container.status == "running":
+                        container.kill()
+
+                    # Log collection
+                    logs_stdout = []
+                    logs_stderr = []
+                    logs_stderr.extend(
+                        container.logs(
+                            stdout=False,
+                            stderr=True,
+                            timestamps=True,
+                            since=datetime2int(self.job_started_at),
+                        )
+                        .decode()
+                        .split("\n")
                     )
-                    .decode()
-                    .split("\n")
-                )
-                logs_stdout.extend(
-                    container.logs(
-                        stdout=True,
-                        stderr=False,
-                        timestamps=True,
-                        since=datetime2int(self.job_started_at),
+                    logs_stdout.extend(
+                        container.logs(
+                            stdout=True,
+                            stderr=False,
+                            timestamps=True,
+                            since=datetime2int(self.job_started_at),
+                        )
+                        .decode()
+                        .split("\n")
                     )
-                    .decode()
-                    .split("\n")
-                )
 
-                # Process logs
-                logs_stdout = [x for x in logs_stdout if len(x) > 0]
-                logs_stderr = [x for x in logs_stderr if len(x) > 0]
-                logs = []
-                for line in logs_stdout + logs_stderr:
-                    date, line = line.split(" ", 1)
-                    date_obj = (
-                        dateutil.parser.parse(date)
-                        .astimezone(datetime.timezone.utc)
-                        .replace(tzinfo=None)
+                    # Process logs
+                    logs_stdout = [x for x in logs_stdout if len(x) > 0]
+                    logs_stderr = [x for x in logs_stderr if len(x) > 0]
+                    logs = []
+                    for line in logs_stdout + logs_stderr:
+                        date, line = line.split(" ", 1)
+                        date_obj = (
+                            dateutil.parser.parse(date)
+                            .astimezone(datetime.timezone.utc)
+                            .replace(tzinfo=None)
+                        )
+                        date = unix_time_millis(date_obj)
+                        logs.append({"timestamp": date, "message": line.strip()})
+                    logs = sorted(logs, key=lambda log: log["timestamp"])
+
+                    # Send to cloudwatch
+                    self.log_stream_name = self._stream_name
+                    self._log_backend.ensure_log_group(self._log_group, None)
+                    self._log_backend.ensure_log_stream(
+                        self._log_group, self.log_stream_name
+                    )
+                    self._log_backend.put_log_events(
+                        self._log_group, self.log_stream_name, logs
                     )
-                    date = unix_time_millis(date_obj)
-                    logs.append({"timestamp": date, "message": line.strip()})
-                logs = sorted(logs, key=lambda log: log["timestamp"])
-
-                # Send to cloudwatch
-                self.log_stream_name = self._stream_name
-                self._log_backend.ensure_log_group(self._log_group, None)
-                self._log_backend.create_log_stream(self._log_group, self._stream_name)
-                self._log_backend.put_log_events(
-                    self._log_group, self._stream_name, logs
-                )
 
-                result = container.wait() or {}
-                exit_code = result.get("StatusCode", 0)
-                self.exit_code = exit_code
-                job_failed = self.stop or exit_code > 0
-                self._mark_stopped(success=not job_failed)
+                    result = container.wait() or {}
+                    exit_code = result.get("StatusCode", 0)
+                    self.exit_code = exit_code
+                    job_failed = self.stop or exit_code > 0
+                    if job_failed:
+                        self._mark_stopped(success=False)
+                        break
 
-            except Exception as err:
-                logger.error(
-                    f"Failed to run AWS Batch container {self.name}. Error {err}"
-                )
-                self._mark_stopped(success=False)
-                container.kill()
-            finally:
-                container.remove()
+                except Exception as err:
+                    logger.error(
+                        f"Failed to run AWS Batch container {self.name}. Error {err}"
+                    )
+                    self._mark_stopped(success=False)
+
+            self._mark_stopped(success=True)
         except Exception as err:
             logger.error(f"Failed to run AWS Batch container {self.name}. Error {err}")
             self._mark_stopped(success=False)
+        finally:
+            for container in containers:
+                container.reload()
+                if container.status == "running":
+                    container.kill()
+                container.remove()
 
     def _mark_stopped(self, success: bool = True) -> None:
+        if self.job_stopped:
+            return
         # Ensure that job_stopped/job_stopped_at-attributes are set first
         # The describe-method needs them immediately when status is set
         self.job_stopped = True
@@ -1437,6 +1536,7 @@ class BatchBackend(BaseBackend):
         tags: Dict[str, str],
         retry_strategy: Dict[str, Any],
         container_properties: Dict[str, Any],
+        node_properties: Dict[str, Any],
         timeout: Dict[str, int],
         platform_capabilities: List[str],
         propagate_tags: bool,
@@ -1457,6 +1557,7 @@ class BatchBackend(BaseBackend):
                 parameters,
                 _type,
                 container_properties,
+                node_properties=node_properties,
                 tags=tags,
                 retry_strategy=retry_strategy,
                 timeout=timeout,
@@ -1467,7 +1568,13 @@ class BatchBackend(BaseBackend):
         else:
             # Make new jobdef
             job_def = job_def.update(
-                parameters, _type, container_properties, retry_strategy, tags, timeout
+                parameters,
+                _type,
+                container_properties,
+                node_properties,
+                retry_strategy,
+                tags,
+                timeout,
             )
 
         self._job_definitions[job_def.arn] = job_def
@@ -1562,7 +1669,11 @@ class BatchBackend(BaseBackend):
 
         result = []
         for key, job in self._jobs.items():
-            if len(job_filter) > 0 and key not in job_filter:
+            if (
+                len(job_filter) > 0
+                and key not in job_filter
+                and job.arn not in job_filter
+            ):
                 continue
 
             result.append(job.describe())
@@ -1570,7 +1681,10 @@ class BatchBackend(BaseBackend):
         return result
 
     def list_jobs(
-        self, job_queue_name: str, job_status: Optional[str] = None
+        self,
+        job_queue_name: str,
+        job_status: Optional[str] = None,
+        filters: Optional[List[Dict[str, Any]]] = None,
     ) -> List[Job]:
         """
         Pagination is not yet implemented
@@ -1598,6 +1712,18 @@ class BatchBackend(BaseBackend):
             if job_status is not None and job.status != job_status:
                 continue
 
+            if filters is not None:
+                matches = True
+                for filt in filters:
+                    name = filt["name"]
+                    values = filt["values"]
+                    if name == "JOB_NAME":
+                        if job.job_name not in values:
+                            matches = False
+                            break
+                if not matches:
+                    continue
+
             jobs.append(job)
 
         return jobs
diff --git a/moto/batch/responses.py b/moto/batch/responses.py
index be4375dd5..1e42b6c76 100644
--- a/moto/batch/responses.py
+++ b/moto/batch/responses.py
@@ -134,6 +134,7 @@ class BatchResponse(BaseResponse):
     # RegisterJobDefinition
     def registerjobdefinition(self) -> str:
         container_properties = self._get_param("containerProperties")
+        node_properties = self._get_param("nodeProperties")
         def_name = self._get_param("jobDefinitionName")
         parameters = self._get_param("parameters")
         tags = self._get_param("tags")
@@ -149,6 +150,7 @@ class BatchResponse(BaseResponse):
             tags=tags,
             retry_strategy=retry_strategy,
             container_properties=container_properties,
+            node_properties=node_properties,
             timeout=timeout,
             platform_capabilities=platform_capabilities,
             propagate_tags=propagate_tags,
@@ -215,8 +217,9 @@ class BatchResponse(BaseResponse):
     def listjobs(self) -> str:
         job_queue = self._get_param("jobQueue")
         job_status = self._get_param("jobStatus")
+        filters = self._get_param("filters")
 
-        jobs = self.batch_backend.list_jobs(job_queue, job_status)
+        jobs = self.batch_backend.list_jobs(job_queue, job_status, filters)
 
         result = {"jobSummaryList": [job.describe_short() for job in jobs]}
         return json.dumps(result)
diff --git a/moto/batch/utils.py b/moto/batch/utils.py
index c565f5636..6df059dca 100644
--- a/moto/batch/utils.py
+++ b/moto/batch/utils.py
@@ -9,6 +9,10 @@ def make_arn_for_job_queue(account_id: str, name: str, region_name: str) -> str:
     return f"arn:aws:batch:{region_name}:{account_id}:job-queue/{name}"
 
 
+def make_arn_for_job(account_id: str, job_id: str, region_name: str) -> str:
+    return f"arn:aws:batch:{region_name}:{account_id}:job/{job_id}"
+
+
 def make_arn_for_task_def(
     account_id: str, name: str, revision: int, region_name: str
 ) -> str:
diff --git a/moto/core/common_types.py b/moto/core/common_types.py
index 7a27c8404..3f04dfbf6 100644
--- a/moto/core/common_types.py
+++ b/moto/core/common_types.py
@@ -1,5 +1,5 @@
-from typing import Dict, Tuple, TypeVar
+from typing import Dict, Tuple, TypeVar, Union
 
 
-TYPE_RESPONSE = Tuple[int, Dict[str, str], str]
+TYPE_RESPONSE = Tuple[int, Dict[str, str], Union[str, bytes]]
 TYPE_IF_NONE = TypeVar("TYPE_IF_NONE")
diff --git a/moto/dynamodb/models/__init__.py b/moto/dynamodb/models/__init__.py
index 854727ee7..9360a4155 100644
--- a/moto/dynamodb/models/__init__.py
+++ b/moto/dynamodb/models/__init__.py
@@ -1422,7 +1422,7 @@ class DynamoDBBackend(BaseBackend):
         else:
             return table.schema
 
-    def get_table(self, table_name):
+    def get_table(self, table_name) -> Table:
         if table_name not in self.tables:
             raise ResourceNotFoundException()
         return self.tables.get(table_name)
diff --git a/moto/dynamodb/responses.py b/moto/dynamodb/responses.py
index 1b4fb83f8..0144c835b 100644
--- a/moto/dynamodb/responses.py
+++ b/moto/dynamodb/responses.py
@@ -13,7 +13,7 @@ from .exceptions import (
     ResourceNotFoundException,
     ConditionalCheckFailed,
 )
-from moto.dynamodb.models import dynamodb_backends, dynamo_json_dump
+from moto.dynamodb.models import dynamodb_backends, dynamo_json_dump, Table
 from moto.utilities.aws_headers import amz_crc32, amzn_request_id
 
 
@@ -67,7 +67,7 @@ def include_consumed_capacity(val=1.0):
     return _inner
 
 
-def get_empty_keys_on_put(field_updates, table):
+def get_empty_keys_on_put(field_updates, table: Table):
     """
     Return the first key-name that has an empty value. None if all keys are filled
     """
@@ -105,6 +105,16 @@ def put_has_empty_attrs(field_updates, table):
     return False
 
 
+def validate_put_has_gsi_keys_set_to_none(item, table: Table) -> None:
+    for gsi in table.global_indexes:
+        for attr in gsi.schema:
+            attr_name = attr["AttributeName"]
+            if attr_name in item and item[attr_name] == {"NULL": True}:
+                raise MockValidationException(
+                    f"One or more parameter values were invalid: Type mismatch for Index Key {attr_name} Expected: S Actual: NULL IndexName: {gsi.name}"
+                )
+
+
 def check_projection_expression(expression):
     if expression.upper() in ReservedKeywords.get_reserved_keywords():
         raise MockValidationException(
@@ -375,15 +385,17 @@ class DynamoHandler(BaseResponse):
         if return_values not in ("ALL_OLD", "NONE"):
             raise MockValidationException("Return values set to invalid value")
 
-        empty_key = get_empty_keys_on_put(item, self.dynamodb_backend.get_table(name))
+        table = self.dynamodb_backend.get_table(name)
+        empty_key = get_empty_keys_on_put(item, table)
         if empty_key:
             raise MockValidationException(
                 f"One or more parameter values were invalid: An AttributeValue may not contain an empty string. Key: {empty_key}"
             )
-        if put_has_empty_attrs(item, self.dynamodb_backend.get_table(name)):
+        if put_has_empty_attrs(item, table):
             raise MockValidationException(
                 "One or more parameter values were invalid: An number set  may not be empty"
             )
+        validate_put_has_gsi_keys_set_to_none(item, table)
 
         overwrite = "Expected" not in self.body
         if not overwrite:
diff --git a/moto/logs/models.py b/moto/logs/models.py
index 581899503..5e241c9c4 100644
--- a/moto/logs/models.py
+++ b/moto/logs/models.py
@@ -649,6 +649,15 @@ class LogsBackend(BaseBackend):
         log_group = self.groups[log_group_name]
         return log_group.create_log_stream(log_stream_name)
 
+    def ensure_log_stream(self, log_group_name: str, log_stream_name: str) -> None:
+        if log_group_name not in self.groups:
+            raise ResourceNotFoundException()
+
+        if log_stream_name in self.groups[log_group_name].streams:
+            return
+
+        self.create_log_stream(log_group_name, log_stream_name)
+
     def delete_log_stream(self, log_group_name, log_stream_name):
         if log_group_name not in self.groups:
             raise ResourceNotFoundException()


Test patch:
diff --git a/tests/test_apigateway/resources/petstore-swagger-v3.yaml b/tests/test_apigateway/resources/petstore-swagger-v3.yaml
new file mode 100644
index 000000000..58d869ae2
--- /dev/null
+++ b/tests/test_apigateway/resources/petstore-swagger-v3.yaml
@@ -0,0 +1,819 @@
+openapi: 3.0.2
+servers:
+  - url: /v3
+info:
+  description: |-
+    This is a sample Pet Store Server based on the OpenAPI 3.0 specification.  You can find out more about
+    Swagger at [http://swagger.io](http://swagger.io). In the third iteration of the pet store, we've switched to the design first approach!
+    You can now help us improve the API whether it's by making changes to the definition itself or to the code.
+    That way, with time, we can improve the API in general, and expose some of the new features in OAS3.
+
+    Some useful links:
+    - [The Pet Store repository](https://github.com/swagger-api/swagger-petstore)
+    - [The source API definition for the Pet Store](https://github.com/swagger-api/swagger-petstore/blob/master/src/main/resources/openapi.yaml)
+  version: 1.0.17
+  title: Swagger Petstore - OpenAPI 3.0
+  termsOfService: 'http://swagger.io/terms/'
+  contact:
+    email: apiteam@swagger.io
+  license:
+    name: Apache 2.0
+    url: 'http://www.apache.org/licenses/LICENSE-2.0.html'
+tags:
+  - name: pet
+    description: Everything about your Pets
+    externalDocs:
+      description: Find out more
+      url: 'http://swagger.io'
+  - name: store
+    description: Access to Petstore orders
+    externalDocs:
+      description: Find out more about our store
+      url: 'http://swagger.io'
+  - name: user
+    description: Operations about user
+paths:
+  /pet:
+    post:
+      tags:
+        - pet
+      summary: Add a new pet to the store
+      description: Add a new pet to the store
+      operationId: addPet
+      responses:
+        '200':
+          description: Successful operation
+          content:
+            application/xml:
+              schema:
+                $ref: '#/components/schemas/Pet'
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Pet'
+        '405':
+          description: Invalid input
+      security:
+        - petstore_auth:
+            - 'write:pets'
+            - 'read:pets'
+      requestBody:
+        description: Create a new pet in the store
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/Pet'
+          application/xml:
+            schema:
+              $ref: '#/components/schemas/Pet'
+          application/x-www-form-urlencoded:
+            schema:
+              $ref: '#/components/schemas/Pet'
+    put:
+      tags:
+        - pet
+      summary: Update an existing pet
+      description: Update an existing pet by Id
+      operationId: updatePet
+      responses:
+        '200':
+          description: Successful operation
+          content:
+            application/xml:
+              schema:
+                $ref: '#/components/schemas/Pet'
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Pet'
+        '400':
+          description: Invalid ID supplied
+        '404':
+          description: Pet not found
+        '405':
+          description: Validation exception
+      security:
+        - petstore_auth:
+            - 'write:pets'
+            - 'read:pets'
+      requestBody:
+        description: Update an existent pet in the store
+        required: true
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/Pet'
+          application/xml:
+            schema:
+              $ref: '#/components/schemas/Pet'
+          application/x-www-form-urlencoded:
+            schema:
+              $ref: '#/components/schemas/Pet'
+  /pet/findByStatus:
+    get:
+      tags:
+        - pet
+      summary: Finds Pets by status
+      description: Multiple status values can be provided with comma separated strings
+      operationId: findPetsByStatus
+      parameters:
+        - name: status
+          in: query
+          description: Status values that need to be considered for filter
+          required: false
+          explode: true
+          schema:
+            type: string
+            enum:
+              - available
+              - pending
+              - sold
+            default: available
+      responses:
+        '200':
+          description: successful operation
+          content:
+            application/xml:
+              schema:
+                type: array
+                items:
+                  $ref: '#/components/schemas/Pet'
+            application/json:
+              schema:
+                type: array
+                items:
+                  $ref: '#/components/schemas/Pet'
+        '400':
+          description: Invalid status value
+      security:
+        - petstore_auth:
+            - 'write:pets'
+            - 'read:pets'
+  /pet/findByTags:
+    get:
+      tags:
+        - pet
+      summary: Finds Pets by tags
+      description: >-
+        Multiple tags can be provided with comma separated strings. Use tag1,
+        tag2, tag3 for testing.
+      operationId: findPetsByTags
+      parameters:
+        - name: tags
+          in: query
+          description: Tags to filter by
+          required: false
+          explode: true
+          schema:
+            type: array
+            items:
+              type: string
+      responses:
+        '200':
+          description: successful operation
+          content:
+            application/xml:
+              schema:
+                type: array
+                items:
+                  $ref: '#/components/schemas/Pet'
+            application/json:
+              schema:
+                type: array
+                items:
+                  $ref: '#/components/schemas/Pet'
+        '400':
+          description: Invalid tag value
+      security:
+        - petstore_auth:
+            - 'write:pets'
+            - 'read:pets'
+  '/pet/{petId}':
+    get:
+      tags:
+        - pet
+      summary: Find pet by ID
+      description: Returns a single pet
+      operationId: getPetById
+      parameters:
+        - name: petId
+          in: path
+          description: ID of pet to return
+          required: true
+          schema:
+            type: integer
+            format: int64
+      responses:
+        '200':
+          description: successful operation
+          content:
+            application/xml:
+              schema:
+                $ref: '#/components/schemas/Pet'
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Pet'
+        '400':
+          description: Invalid ID supplied
+        '404':
+          description: Pet not found
+      security:
+        - api_key: []
+        - petstore_auth:
+            - 'write:pets'
+            - 'read:pets'
+    post:
+      tags:
+        - pet
+      summary: Updates a pet in the store with form data
+      description: ''
+      operationId: updatePetWithForm
+      parameters:
+        - name: petId
+          in: path
+          description: ID of pet that needs to be updated
+          required: true
+          schema:
+            type: integer
+            format: int64
+        - name: name
+          in: query
+          description: Name of pet that needs to be updated
+          schema:
+            type: string
+        - name: status
+          in: query
+          description: Status of pet that needs to be updated
+          schema:
+            type: string
+      responses:
+        '405':
+          description: Invalid input
+      security:
+        - petstore_auth:
+            - 'write:pets'
+            - 'read:pets'
+    delete:
+      tags:
+        - pet
+      summary: Deletes a pet
+      description: ''
+      operationId: deletePet
+      parameters:
+        - name: api_key
+          in: header
+          description: ''
+          required: false
+          schema:
+            type: string
+        - name: petId
+          in: path
+          description: Pet id to delete
+          required: true
+          schema:
+            type: integer
+            format: int64
+      responses:
+        '400':
+          description: Invalid pet value
+      security:
+        - petstore_auth:
+            - 'write:pets'
+            - 'read:pets'
+  '/pet/{petId}/uploadImage':
+    post:
+      tags:
+        - pet
+      summary: uploads an image
+      description: ''
+      operationId: uploadFile
+      parameters:
+        - name: petId
+          in: path
+          description: ID of pet to update
+          required: true
+          schema:
+            type: integer
+            format: int64
+        - name: additionalMetadata
+          in: query
+          description: Additional Metadata
+          required: false
+          schema:
+            type: string
+      responses:
+        '200':
+          description: successful operation
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/ApiResponse'
+      security:
+        - petstore_auth:
+            - 'write:pets'
+            - 'read:pets'
+      requestBody:
+        content:
+          application/octet-stream:
+            schema:
+              type: string
+              format: binary
+  /store/inventory:
+    get:
+      tags:
+        - store
+      summary: Returns pet inventories by status
+      description: Returns a map of status codes to quantities
+      operationId: getInventory
+      x-swagger-router-controller: OrderController
+      responses:
+        '200':
+          description: successful operation
+          content:
+            application/json:
+              schema:
+                type: object
+                additionalProperties:
+                  type: integer
+                  format: int32
+      security:
+        - api_key: []
+  /store/order:
+    post:
+      tags:
+        - store
+      summary: Place an order for a pet
+      description: Place a new order in the store
+      operationId: placeOrder
+      x-swagger-router-controller: OrderController
+      responses:
+        '200':
+          description: successful operation
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Order'
+        '405':
+          description: Invalid input
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/Order'
+          application/xml:
+            schema:
+              $ref: '#/components/schemas/Order'
+          application/x-www-form-urlencoded:
+            schema:
+              $ref: '#/components/schemas/Order'
+  '/store/order/{orderId}':
+    get:
+      tags:
+        - store
+      summary: Find purchase order by ID
+      x-swagger-router-controller: OrderController
+      description: >-
+        For valid response try integer IDs with value <= 5 or > 10. Other values
+        will generate exceptions.
+      operationId: getOrderById
+      parameters:
+        - name: orderId
+          in: path
+          description: ID of order that needs to be fetched
+          required: true
+          schema:
+            type: integer
+            format: int64
+      responses:
+        '200':
+          description: successful operation
+          content:
+            application/xml:
+              schema:
+                $ref: '#/components/schemas/Order'
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Order'
+        '400':
+          description: Invalid ID supplied
+        '404':
+          description: Order not found
+    delete:
+      tags:
+        - store
+      summary: Delete purchase order by ID
+      x-swagger-router-controller: OrderController
+      description: >-
+        For valid response try integer IDs with value < 1000. Anything above
+        1000 or nonintegers will generate API errors
+      operationId: deleteOrder
+      parameters:
+        - name: orderId
+          in: path
+          description: ID of the order that needs to be deleted
+          required: true
+          schema:
+            type: integer
+            format: int64
+      responses:
+        '400':
+          description: Invalid ID supplied
+        '404':
+          description: Order not found
+  /user:
+    post:
+      tags:
+        - user
+      summary: Create user
+      description: This can only be done by the logged in user.
+      operationId: createUser
+      responses:
+        default:
+          description: successful operation
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/User'
+            application/xml:
+              schema:
+                $ref: '#/components/schemas/User'
+      requestBody:
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/User'
+          application/xml:
+            schema:
+              $ref: '#/components/schemas/User'
+          application/x-www-form-urlencoded:
+            schema:
+              $ref: '#/components/schemas/User'
+        description: Created user object
+  /user/createWithList:
+    post:
+      tags:
+        - user
+      summary: Creates list of users with given input array
+      description: 'Creates list of users with given input array'
+      x-swagger-router-controller: UserController
+      operationId: createUsersWithListInput
+      responses:
+        '200':
+          description: Successful operation
+          content:
+            application/xml:
+              schema:
+                $ref: '#/components/schemas/User'
+            application/json:
+              schema:
+                $ref: '#/components/schemas/User'
+        default:
+          description: successful operation
+      requestBody:
+        content:
+          application/json:
+            schema:
+              type: array
+              items:
+                $ref: '#/components/schemas/User'
+  /user/login:
+    get:
+      tags:
+        - user
+      summary: Logs user into the system
+      description: ''
+      operationId: loginUser
+      parameters:
+        - name: username
+          in: query
+          description: The user name for login
+          required: false
+          schema:
+            type: string
+        - name: password
+          in: query
+          description: The password for login in clear text
+          required: false
+          schema:
+            type: string
+      responses:
+        '200':
+          description: successful operation
+          headers:
+            X-Rate-Limit:
+              description: calls per hour allowed by the user
+              schema:
+                type: integer
+                format: int32
+            X-Expires-After:
+              description: date in UTC when token expires
+              schema:
+                type: string
+                format: date-time
+          content:
+            application/xml:
+              schema:
+                type: string
+            application/json:
+              schema:
+                type: string
+        '400':
+          description: Invalid username/password supplied
+  /user/logout:
+    get:
+      tags:
+        - user
+      summary: Logs out current logged in user session
+      description: ''
+      operationId: logoutUser
+      parameters: []
+      responses:
+        default:
+          description: successful operation
+  '/user/{username}':
+    get:
+      tags:
+        - user
+      summary: Get user by user name
+      description: ''
+      operationId: getUserByName
+      parameters:
+        - name: username
+          in: path
+          description: 'The name that needs to be fetched. Use user1 for testing. '
+          required: true
+          schema:
+            type: string
+      responses:
+        '200':
+          description: successful operation
+          content:
+            application/xml:
+              schema:
+                $ref: '#/components/schemas/User'
+            application/json:
+              schema:
+                $ref: '#/components/schemas/User'
+        '400':
+          description: Invalid username supplied
+        '404':
+          description: User not found
+    put:
+      tags:
+        - user
+      summary: Update user
+      x-swagger-router-controller: UserController
+      description: This can only be done by the logged in user.
+      operationId: updateUser
+      parameters:
+        - name: username
+          in: path
+          description: name that need to be deleted
+          required: true
+          schema:
+            type: string
+      responses:
+        default:
+          description: successful operation
+      requestBody:
+        description: Update an existent user in the store
+        content:
+          application/json:
+            schema:
+              $ref: '#/components/schemas/User'
+          application/xml:
+            schema:
+              $ref: '#/components/schemas/User'
+          application/x-www-form-urlencoded:
+            schema:
+              $ref: '#/components/schemas/User'
+    delete:
+      tags:
+        - user
+      summary: Delete user
+      description: This can only be done by the logged in user.
+      operationId: deleteUser
+      parameters:
+        - name: username
+          in: path
+          description: The name that needs to be deleted
+          required: true
+          schema:
+            type: string
+      responses:
+        '400':
+          description: Invalid username supplied
+        '404':
+          description: User not found
+externalDocs:
+  description: Find out more about Swagger
+  url: 'http://swagger.io'
+components:
+  schemas:
+    Order:
+      x-swagger-router-model: io.swagger.petstore.model.Order
+      properties:
+        id:
+          type: integer
+          format: int64
+          example: 10
+        petId:
+          type: integer
+          format: int64
+          example: 198772
+        quantity:
+          type: integer
+          format: int32
+          example: 7
+        shipDate:
+          type: string
+          format: date-time
+        status:
+          type: string
+          description: Order Status
+          enum:
+            - placed
+            - approved
+            - delivered
+          example: approved
+        complete:
+          type: boolean
+      xml:
+        name: order
+      type: object
+    Customer:
+      properties:
+        id:
+          type: integer
+          format: int64
+          example: 100000
+        username:
+          type: string
+          example: fehguy
+        address:
+          type: array
+          items:
+            $ref: '#/components/schemas/Address'
+          xml:
+            wrapped: true
+            name: addresses
+      xml:
+        name: customer
+      type: object
+    Address:
+      properties:
+        street:
+          type: string
+          example: 437 Lytton
+        city:
+          type: string
+          example: Palo Alto
+        state:
+          type: string
+          example: CA
+        zip:
+          type: string
+          example: 94301
+      xml:
+        name: address
+      type: object
+    Category:
+      x-swagger-router-model: io.swagger.petstore.model.Category
+      properties:
+        id:
+          type: integer
+          format: int64
+          example: 1
+        name:
+          type: string
+          example: Dogs
+      xml:
+        name: category
+      type: object
+    User:
+      x-swagger-router-model: io.swagger.petstore.model.User
+      properties:
+        id:
+          type: integer
+          format: int64
+          example: 10
+        username:
+          type: string
+          example: theUser
+        firstName:
+          type: string
+          example: John
+        lastName:
+          type: string
+          example: James
+        email:
+          type: string
+          example: john@email.com
+        password:
+          type: string
+          example: 12345
+        phone:
+          type: string
+          example: 12345
+        userStatus:
+          type: integer
+          format: int32
+          example: 1
+          description: User Status
+      xml:
+        name: user
+      type: object
+    Tag:
+      x-swagger-router-model: io.swagger.petstore.model.Tag
+      properties:
+        id:
+          type: integer
+          format: int64
+        name:
+          type: string
+      xml:
+        name: tag
+      type: object
+    Pet:
+      x-swagger-router-model: io.swagger.petstore.model.Pet
+      required:
+        - name
+        - photoUrls
+      properties:
+        id:
+          type: integer
+          format: int64
+          example: 10
+        name:
+          type: string
+          example: doggie
+        category:
+          $ref: '#/components/schemas/Category'
+        photoUrls:
+          type: array
+          xml:
+            wrapped: true
+          items:
+            type: string
+            xml:
+              name: photoUrl
+        tags:
+          type: array
+          xml:
+            wrapped: true
+          items:
+            $ref: '#/components/schemas/Tag'
+            xml:
+              name: tag
+        status:
+          type: string
+          description: pet status in the store
+          enum:
+            - available
+            - pending
+            - sold
+      xml:
+        name: pet
+      type: object
+    ApiResponse:
+      properties:
+        code:
+          type: integer
+          format: int32
+        type:
+          type: string
+        message:
+          type: string
+      xml:
+        name: '##default'
+      type: object
+  requestBodies:
+    Pet:
+      content:
+        application/json:
+          schema:
+            $ref: '#/components/schemas/Pet'
+        application/xml:
+          schema:
+            $ref: '#/components/schemas/Pet'
+      description: Pet object that needs to be added to the store
+    UserArray:
+      content:
+        application/json:
+          schema:
+            type: array
+            items:
+              $ref: '#/components/schemas/User'
+      description: List of user object
+  securitySchemes:
+    petstore_auth:
+      type: oauth2
+      flows:
+        implicit:
+          authorizationUrl: 'https://petstore.swagger.io/oauth/authorize'
+          scopes:
+            'write:pets': modify pets in your account
+            'read:pets': read your pets
+    api_key:
+      type: apiKey
+      name: api_key
+      in: header
diff --git a/tests/test_apigateway/resources/test_deep_api.yaml b/tests/test_apigateway/resources/test_deep_api.yaml
new file mode 100644
index 000000000..778e1843d
--- /dev/null
+++ b/tests/test_apigateway/resources/test_deep_api.yaml
@@ -0,0 +1,50 @@
+openapi: 3.0.0
+info:
+  description: description
+  version: '1'
+  title: doc
+paths:
+  /test/some/deep/path:
+    post:
+      description: Method description.
+      responses:
+        '200':
+          description: 200 response
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/Empty'
+
+x-amazon-apigateway-documentation:
+  version: 1.0.3
+  documentationParts:
+    - location:
+        type: API
+      properties:
+        description: API description
+        info:
+          description: API info description 4
+          version: API info version 3
+    - location:
+        type: METHOD
+        method: GET
+      properties:
+        description: Method description.
+    - location:
+        type: MODEL
+        name: Empty
+      properties:
+        title: Empty Schema
+    - location:
+        type: RESPONSE
+        method: GET
+        statusCode: '200'
+      properties:
+        description: 200 response
+servers:
+  - url: /
+components:
+  schemas:
+    Empty:
+      type: object
+      title: Empty Schema
diff --git a/tests/test_apigateway/test_apigateway_export.py b/tests/test_apigateway/test_apigateway_export.py
new file mode 100644
index 000000000..90b74ec70
--- /dev/null
+++ b/tests/test_apigateway/test_apigateway_export.py
@@ -0,0 +1,81 @@
+import boto3
+import json
+import os
+import pytest
+
+from botocore.exceptions import ClientError
+from moto import mock_apigateway
+
+
+@mock_apigateway
+def test_import_rest_api__api_is_created():
+    client = boto3.client("apigateway", region_name="us-west-2")
+
+    path = os.path.dirname(os.path.abspath(__file__))
+    with open(path + "/resources/petstore-swagger-v3.yaml", "rb") as api_json:
+        response = client.import_rest_api(body=api_json.read())
+        api_id = response["id"]
+
+    root_id = client.get_resources(restApiId=api_id)["items"][4]["id"]
+    client.get_method(restApiId=api_id, resourceId=root_id, httpMethod="POST")
+
+    client.put_integration(
+        restApiId=api_id, resourceId=root_id, httpMethod="POST", type="MOCK"
+    )
+    client.put_integration_response(
+        restApiId=api_id, resourceId=root_id, httpMethod="POST", statusCode="200"
+    )
+
+    deployment_id = client.create_deployment(restApiId=api_id, stageName="dep")["id"]
+
+    client.create_stage(restApiId=api_id, stageName="stg", deploymentId=deployment_id)
+
+    resp = client.get_export(restApiId=api_id, stageName="stg", exportType="swagger")
+    resp.should.have.key("contentType").equals("application/octet-stream")
+    resp.should.have.key("contentDisposition").match('attachment; filename="swagger')
+
+    body = json.loads(resp["body"].read().decode("utf-8"))
+
+    body["info"].should.have.key("title").equals("Swagger Petstore - OpenAPI 3.0")
+    body["paths"].should.have.length_of(15)
+
+    body["paths"]["/pet/{petId}"].should.have.length_of(3)
+    set(body["paths"]["/pet/{petId}"].keys()).should.equal({"DELETE", "GET", "POST"})
+
+
+@mock_apigateway
+def test_export_api__unknown_api():
+    client = boto3.client("apigateway", region_name="us-west-2")
+
+    with pytest.raises(ClientError) as exc:
+        client.get_export(restApiId="unknown", stageName="l", exportType="a")
+    err = exc.value.response["Error"]
+    err["Code"].should.equal("NotFoundException")
+    err["Message"].should.equal("Invalid stage identifier specified")
+
+
+@mock_apigateway
+def test_export_api__unknown_export_type():
+    client = boto3.client("apigateway", region_name="us-east-1")
+
+    path = os.path.dirname(os.path.abspath(__file__))
+    with open(path + "/resources/petstore-swagger-v3.yaml", "rb") as api_json:
+        response = client.import_rest_api(body=api_json.read())
+        api_id = response["id"]
+
+    root_id = client.get_resources(restApiId=api_id)["items"][4]["id"]
+    client.get_method(restApiId=api_id, resourceId=root_id, httpMethod="POST")
+
+    client.put_integration(
+        restApiId=api_id, resourceId=root_id, httpMethod="POST", type="MOCK"
+    )
+    client.put_integration_response(
+        restApiId=api_id, resourceId=root_id, httpMethod="POST", statusCode="200"
+    )
+    client.create_deployment(restApiId=api_id, stageName="dep")
+
+    with pytest.raises(ClientError) as exc:
+        client.get_export(restApiId=api_id, stageName="dep", exportType="a")
+    err = exc.value.response["Error"]
+    err["Code"].should.equal("BadRequestException")
+    err["Message"].should.equal("No API exporter for type 'a'")
diff --git a/tests/test_apigateway/test_apigateway_importrestapi.py b/tests/test_apigateway/test_apigateway_importrestapi.py
index 12d004538..0ecd35e98 100644
--- a/tests/test_apigateway/test_apigateway_importrestapi.py
+++ b/tests/test_apigateway/test_apigateway_importrestapi.py
@@ -27,6 +27,25 @@ def test_import_rest_api__api_is_created():
     )
 
 
+@mock_apigateway
+def test_import_rest_api__nested_api():
+    client = boto3.client("apigateway", region_name="us-west-2")
+
+    path = os.path.dirname(os.path.abspath(__file__))
+    with open(path + "/resources/test_deep_api.yaml", "rb") as api_json:
+        response = client.import_rest_api(body=api_json.read())
+
+    resources = client.get_resources(restApiId=response["id"])["items"]
+    resources.should.have.length_of(5)
+
+    paths = [res["path"] for res in resources]
+    paths.should.contain("/")
+    paths.should.contain("/test")
+    paths.should.contain("/test/some")
+    paths.should.contain("/test/some/deep")
+    paths.should.contain("/test/some/deep/path")
+
+
 @mock_apigateway
 def test_import_rest_api__invalid_api_creates_nothing():
     client = boto3.client("apigateway", region_name="us-west-2")
diff --git a/tests/test_batch/test_batch_jobs.py b/tests/test_batch/test_batch_jobs.py
index 05d93898f..a521810c2 100644
--- a/tests/test_batch/test_batch_jobs.py
+++ b/tests/test_batch/test_batch_jobs.py
@@ -151,6 +151,72 @@ def test_submit_job():
     attempt.should.have.key("stoppedAt").equals(stopped_at)
 
 
+@mock_logs
+@mock_ec2
+@mock_ecs
+@mock_iam
+@mock_batch
+@pytest.mark.network
+def test_submit_job_multinode():
+    ec2_client, iam_client, _, logs_client, batch_client = _get_clients()
+    _, _, _, iam_arn = _setup(ec2_client, iam_client)
+    start_time_milliseconds = time.time() * 1000
+
+    job_def_name = str(uuid4())[0:6]
+    commands = ["echo", "hello"]
+    job_def_arn, queue_arn = prepare_multinode_job(
+        batch_client, commands, iam_arn, job_def_name
+    )
+
+    resp = batch_client.submit_job(
+        jobName=str(uuid4())[0:6], jobQueue=queue_arn, jobDefinition=job_def_arn
+    )
+    job_id = resp["jobId"]
+
+    # Test that describe_jobs() returns 'createdAt'
+    # github.com/getmoto/moto/issues/4364
+    resp = batch_client.describe_jobs(jobs=[job_id])
+    created_at = resp["jobs"][0]["createdAt"]
+    created_at.should.be.greater_than(start_time_milliseconds)
+
+    _wait_for_job_status(batch_client, job_id, "SUCCEEDED")
+
+    resp = logs_client.describe_log_streams(
+        logGroupName="/aws/batch/job", logStreamNamePrefix=job_def_name
+    )
+    resp["logStreams"].should.have.length_of(1)
+    ls_name = resp["logStreams"][0]["logStreamName"]
+
+    resp = logs_client.get_log_events(
+        logGroupName="/aws/batch/job", logStreamName=ls_name
+    )
+    [event["message"] for event in resp["events"]].should.equal(["hello", "hello"])
+
+    # Test that describe_jobs() returns timestamps in milliseconds
+    # github.com/getmoto/moto/issues/4364
+    job = batch_client.describe_jobs(jobs=[job_id])["jobs"][0]
+    created_at = job["createdAt"]
+    started_at = job["startedAt"]
+    stopped_at = job["stoppedAt"]
+
+    created_at.should.be.greater_than(start_time_milliseconds)
+    started_at.should.be.greater_than(start_time_milliseconds)
+    stopped_at.should.be.greater_than(start_time_milliseconds)
+
+    # Verify we track attempts
+    job.should.have.key("attempts").length_of(1)
+    attempt = job["attempts"][0]
+    attempt.should.have.key("container")
+    attempt["container"].should.have.key("containerInstanceArn")
+    attempt["container"].should.have.key("logStreamName").equals(
+        job["container"]["logStreamName"]
+    )
+    attempt["container"].should.have.key("networkInterfaces")
+    attempt["container"].should.have.key("taskArn")
+    attempt.should.have.key("startedAt").equals(started_at)
+    attempt.should.have.key("stoppedAt").equals(stopped_at)
+
+
 @mock_logs
 @mock_ec2
 @mock_ecs
@@ -205,6 +271,18 @@ def test_list_jobs():
         job.should.have.key("stoppedAt")
         job.should.have.key("container").should.have.key("exitCode").equals(0)
 
+    filtered_jobs = batch_client.list_jobs(
+        jobQueue=queue_arn,
+        filters=[
+            {
+                "name": "JOB_NAME",
+                "values": ["test2"],
+            }
+        ],
+    )["jobSummaryList"]
+    filtered_jobs.should.have.length_of(1)
+    filtered_jobs[0]["jobName"].should.equal("test2")
+
 
 @mock_logs
 @mock_ec2
@@ -772,6 +850,51 @@ def prepare_job(batch_client, commands, iam_arn, job_def_name):
     return job_def_arn, queue_arn
 
 
+def prepare_multinode_job(batch_client, commands, iam_arn, job_def_name):
+    compute_name = str(uuid4())[0:6]
+    resp = batch_client.create_compute_environment(
+        computeEnvironmentName=compute_name,
+        type="UNMANAGED",
+        state="ENABLED",
+        serviceRole=iam_arn,
+    )
+    arn = resp["computeEnvironmentArn"]
+
+    resp = batch_client.create_job_queue(
+        jobQueueName=str(uuid4())[0:6],
+        state="ENABLED",
+        priority=123,
+        computeEnvironmentOrder=[{"order": 123, "computeEnvironment": arn}],
+    )
+    queue_arn = resp["jobQueueArn"]
+    container = {
+        "image": "busybox:latest",
+        "vcpus": 1,
+        "memory": 128,
+        "command": commands,
+    }
+    resp = batch_client.register_job_definition(
+        jobDefinitionName=job_def_name,
+        type="multinode",
+        nodeProperties={
+            "mainNode": 0,
+            "numNodes": 2,
+            "nodeRangeProperties": [
+                {
+                    "container": container,
+                    "targetNodes": "0",
+                },
+                {
+                    "container": container,
+                    "targetNodes": "1",
+                },
+            ],
+        },
+    )
+    job_def_arn = resp["jobDefinitionArn"]
+    return job_def_arn, queue_arn
+
+
 @mock_batch
 def test_update_job_definition():
     _, _, _, _, batch_client = _get_clients()
diff --git a/tests/test_dynamodb/exceptions/test_dynamodb_exceptions.py b/tests/test_dynamodb/exceptions/test_dynamodb_exceptions.py
index 6271f7678..ebb7c13cc 100644
--- a/tests/test_dynamodb/exceptions/test_dynamodb_exceptions.py
+++ b/tests/test_dynamodb/exceptions/test_dynamodb_exceptions.py
@@ -899,3 +899,43 @@ def test_put_item__string_as_integer_value():
     err = exc.value.response["Error"]
     err["Code"].should.equal("SerializationException")
     err["Message"].should.equal("Start of structure or map found where not expected")
+
+
+@mock_dynamodb
+def test_gsi_key_cannot_be_empty():
+    dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
+    hello_index = {
+        "IndexName": "hello-index",
+        "KeySchema": [{"AttributeName": "hello", "KeyType": "HASH"}],
+        "Projection": {"ProjectionType": "ALL"},
+    }
+    table_name = "lilja-test"
+
+    # Let's create a table with [id: str, hello: str], with an index to hello
+    dynamodb.create_table(
+        TableName=table_name,
+        KeySchema=[
+            {"AttributeName": "id", "KeyType": "HASH"},
+        ],
+        AttributeDefinitions=[
+            {"AttributeName": "id", "AttributeType": "S"},
+            {"AttributeName": "hello", "AttributeType": "S"},
+        ],
+        GlobalSecondaryIndexes=[hello_index],
+        BillingMode="PAY_PER_REQUEST",
+    )
+
+    table = dynamodb.Table(table_name)
+    with pytest.raises(ClientError) as exc:
+        table.put_item(
+            TableName=table_name,
+            Item={
+                "id": "woop",
+                "hello": None,
+            },
+        )
+    err = exc.value.response["Error"]
+    err["Code"].should.equal("ValidationException")
+    err["Message"].should.equal(
+        "One or more parameter values were invalid: Type mismatch for Index Key hello Expected: S Actual: NULL IndexName: hello-index"
+    )