We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Bug Report: Glue search_tables with Filters and result tables more than 100 items
<!--- Provide a general summary of the issue in the Title above -->
<!--- Look through existing open and closed issues to see if someone has reported the issue before -->

There is a bug while applying Filters for the database which contains more than 100 items. Since glue returns 100 items per page and to move to the next page we need to specify `NextToken`.
I have 138 tables, which means I will be iterating 2 times over the result.

The filter:
```python
{
        'Key': 'DatabaseName',
        'Value': glue_database_name
}
```

Every time I run the code I get different results: the length of the list is always the same - 138. However, the length of the set is always different. It ranges from 1 to 30.
I run my check over 10 times.

I took look at the documentation and found a proper parameter `MaxResults` for further checking. Since I know precisely desired table count, I put it as 150 and the issue has totally gone.

## Expected Behavior
Get the exact same result for filtered tables.

## Current Behavior
Query result from [`self._glue.search_tables(**kwargs)`](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/glue_extractor.py#L78) contains duplicates

## Possible Solution
I'm not sure, but I think for the next (second) iteration (page, which contains up to 100 items) we are using a new `NextToken` with previous filters. Maybe the problem lies here.

## Steps to Reproduce
1. Have more than 100 glue tables in a single DB in AWS
2. Query it using the abovementioned `DatabaseName` filter
3. Observe duplicates in the list

## Hot-fix
1. Add `MaxResults` to [`kwargs`](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/glue_extractor.py#L80) that is more than your actual size of overall tables
2. Observe a proper behavior

## Context
Q: How has this issue affected you?
A: It affects our production system

## Your Environment
```
amundsen-databuilder==4.5.3
amundsen-gremlin==0.0.9
Flask==1.1.4
gremlinpython==3.4.9
requests-aws4auth==1.1.1
typing-extensions==3.10.0
overrides==6.1.0
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `databuilder/databuilder/extractor/glue_extractor.py`
Content:
```
1 # Copyright Contributors to the Amundsen project.
2 # SPDX-License-Identifier: Apache-2.0
3 
4 from typing import (
5     Any, Dict, Iterator, List, Union,
6 )
7 
8 import boto3
9 from pyhocon import ConfigFactory, ConfigTree
10 
11 from databuilder.extractor.base_extractor import Extractor
12 from databuilder.models.table_metadata import ColumnMetadata, TableMetadata
13 
14 
15 class GlueExtractor(Extractor):
16     """
17     Extracts tables and columns metadata from AWS Glue metastore
18     """
19 
20     CLUSTER_KEY = 'cluster'
21     FILTER_KEY = 'filters'
22     DEFAULT_CONFIG = ConfigFactory.from_dict({CLUSTER_KEY: 'gold', FILTER_KEY: None})
23 
24     def init(self, conf: ConfigTree) -> None:
25         conf = conf.with_fallback(GlueExtractor.DEFAULT_CONFIG)
26         self._cluster = conf.get_string(GlueExtractor.CLUSTER_KEY)
27         self._filters = conf.get(GlueExtractor.FILTER_KEY)
28         self._glue = boto3.client('glue')
29         self._extract_iter: Union[None, Iterator] = None
30 
31     def extract(self) -> Union[TableMetadata, None]:
32         if not self._extract_iter:
33             self._extract_iter = self._get_extract_iter()
34         try:
35             return next(self._extract_iter)
36         except StopIteration:
37             return None
38 
39     def get_scope(self) -> str:
40         return 'extractor.glue'
41 
42     def _get_extract_iter(self) -> Iterator[TableMetadata]:
43         """
44         It gets all tables and yields TableMetadata
45         :return:
46         """
47         for row in self._get_raw_extract_iter():
48             columns, i = [], 0
49 
50             for column in row['StorageDescriptor']['Columns'] \
51                     + row.get('PartitionKeys', []):
52                 columns.append(ColumnMetadata(
53                     column['Name'],
54                     column['Comment'] if 'Comment' in column else None,
55                     column['Type'],
56                     i
57                 ))
58                 i += 1
59 
60             yield TableMetadata(
61                 'glue',
62                 self._cluster,
63                 row['DatabaseName'],
64                 row['Name'],
65                 row.get('Description') or row.get('Parameters', {}).get('comment'),
66                 columns,
67                 row.get('TableType') == 'VIRTUAL_VIEW',
68             )
69 
70     def _get_raw_extract_iter(self) -> Iterator[Dict[str, Any]]:
71         """
72         Provides iterator of results row from glue client
73         :return:
74         """
75         tables = self._search_tables()
76         return iter(tables)
77 
78     def _search_tables(self) -> List[Dict[str, Any]]:
79         tables = []
80         kwargs = {}
81         if self._filters is not None:
82             kwargs['Filters'] = self._filters
83         data = self._glue.search_tables(**kwargs)
84         tables += data['TableList']
85         while 'NextToken' in data:
86             token = data['NextToken']
87             kwargs['NextToken'] = token
88             data = self._glue.search_tables(**kwargs)
89             tables += data['TableList']
90         return tables
91
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/databuilder/databuilder/extractor/glue_extractor.py b/databuilder/databuilder/extractor/glue_extractor.py
--- a/databuilder/databuilder/extractor/glue_extractor.py
+++ b/databuilder/databuilder/extractor/glue_extractor.py
@@ -19,12 +19,14 @@
 
     CLUSTER_KEY = 'cluster'
     FILTER_KEY = 'filters'
-    DEFAULT_CONFIG = ConfigFactory.from_dict({CLUSTER_KEY: 'gold', FILTER_KEY: None})
+    MAX_RESULTS_KEY = 'max_results'
+    DEFAULT_CONFIG = ConfigFactory.from_dict({CLUSTER_KEY: 'gold', FILTER_KEY: None, MAX_RESULTS_KEY: 500})
 
     def init(self, conf: ConfigTree) -> None:
         conf = conf.with_fallback(GlueExtractor.DEFAULT_CONFIG)
         self._cluster = conf.get_string(GlueExtractor.CLUSTER_KEY)
         self._filters = conf.get(GlueExtractor.FILTER_KEY)
+        self._max_results = conf.get(GlueExtractor.MAX_RESULTS_KEY)
         self._glue = boto3.client('glue')
         self._extract_iter: Union[None, Iterator] = None
 
@@ -80,6 +82,7 @@
         kwargs = {}
         if self._filters is not None:
             kwargs['Filters'] = self._filters
+            kwargs['MaxResults'] = self._max_results
         data = self._glue.search_tables(**kwargs)
         tables += data['TableList']
         while 'NextToken' in data:

{"golden_diff": "diff --git a/databuilder/databuilder/extractor/glue_extractor.py b/databuilder/databuilder/extractor/glue_extractor.py\n--- a/databuilder/databuilder/extractor/glue_extractor.py\n+++ b/databuilder/databuilder/extractor/glue_extractor.py\n@@ -19,12 +19,14 @@\n \n     CLUSTER_KEY = 'cluster'\n     FILTER_KEY = 'filters'\n-    DEFAULT_CONFIG = ConfigFactory.from_dict({CLUSTER_KEY: 'gold', FILTER_KEY: None})\n+    MAX_RESULTS_KEY = 'max_results'\n+    DEFAULT_CONFIG = ConfigFactory.from_dict({CLUSTER_KEY: 'gold', FILTER_KEY: None, MAX_RESULTS_KEY: 500})\n \n     def init(self, conf: ConfigTree) -> None:\n         conf = conf.with_fallback(GlueExtractor.DEFAULT_CONFIG)\n         self._cluster = conf.get_string(GlueExtractor.CLUSTER_KEY)\n         self._filters = conf.get(GlueExtractor.FILTER_KEY)\n+        self._max_results = conf.get(GlueExtractor.MAX_RESULTS_KEY)\n         self._glue = boto3.client('glue')\n         self._extract_iter: Union[None, Iterator] = None\n \n@@ -80,6 +82,7 @@\n         kwargs = {}\n         if self._filters is not None:\n             kwargs['Filters'] = self._filters\n+            kwargs['MaxResults'] = self._max_results\n         data = self._glue.search_tables(**kwargs)\n         tables += data['TableList']\n         while 'NextToken' in data:\n", "issue": "Bug Report: Glue search_tables with Filters and result tables more than 100 items\n<!--- Provide a general summary of the issue in the Title above -->\r\n<!--- Look through existing open and closed issues to see if someone has reported the issue before -->\r\n\r\nThere is a bug while applying Filters for the database which contains more than 100 items. Since glue returns 100 items per page and to move to the next page we need to specify `NextToken`.\r\nI have 138 tables, which means I will be iterating 2 times over the result.\r\n\r\nThe filter:\r\n```python\r\n{\r\n        'Key': 'DatabaseName',\r\n        'Value': glue_database_name\r\n}\r\n```\r\n\r\nEvery time I run the code I get different results: the length of the list is always the same - 138. However, the length of the set is always different. It ranges from 1 to 30.\r\nI run my check over 10 times.\r\n\r\nI took look at the documentation and found a proper parameter `MaxResults` for further checking. Since I know precisely desired table count, I put it as 150 and the issue has totally gone.\r\n\r\n## Expected Behavior\r\nGet the exact same result for filtered tables.\r\n\r\n## Current Behavior\r\nQuery result from [`self._glue.search_tables(**kwargs)`](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/glue_extractor.py#L78) contains duplicates\r\n\r\n## Possible Solution\r\nI'm not sure, but I think for the next (second) iteration (page, which contains up to 100 items) we are using a new `NextToken` with previous filters. Maybe the problem lies here.\r\n\r\n## Steps to Reproduce\r\n1. Have more than 100 glue tables in a single DB in AWS\r\n2. Query it using the abovementioned `DatabaseName` filter\r\n3. Observe duplicates in the list\r\n\r\n## Hot-fix\r\n1. Add `MaxResults` to [`kwargs`](https://github.com/amundsen-io/amundsen/blob/main/databuilder/databuilder/extractor/glue_extractor.py#L80) that is more than your actual size of overall tables\r\n2. Observe a proper behavior\r\n\r\n## Context\r\nQ: How has this issue affected you?\r\nA: It affects our production system\r\n\r\n## Your Environment\r\n```\r\namundsen-databuilder==4.5.3\r\namundsen-gremlin==0.0.9\r\nFlask==1.1.4\r\ngremlinpython==3.4.9\r\nrequests-aws4auth==1.1.1\r\ntyping-extensions==3.10.0\r\noverrides==6.1.0\r\n```\n", "before_files": [{"content": "# Copyright Contributors to the Amundsen project.\n# SPDX-License-Identifier: Apache-2.0\n\nfrom typing import (\n    Any, Dict, Iterator, List, Union,\n)\n\nimport boto3\nfrom pyhocon import ConfigFactory, ConfigTree\n\nfrom databuilder.extractor.base_extractor import Extractor\nfrom databuilder.models.table_metadata import ColumnMetadata, TableMetadata\n\n\nclass GlueExtractor(Extractor):\n    \"\"\"\n    Extracts tables and columns metadata from AWS Glue metastore\n    \"\"\"\n\n    CLUSTER_KEY = 'cluster'\n    FILTER_KEY = 'filters'\n    DEFAULT_CONFIG = ConfigFactory.from_dict({CLUSTER_KEY: 'gold', FILTER_KEY: None})\n\n    def init(self, conf: ConfigTree) -> None:\n        conf = conf.with_fallback(GlueExtractor.DEFAULT_CONFIG)\n        self._cluster = conf.get_string(GlueExtractor.CLUSTER_KEY)\n        self._filters = conf.get(GlueExtractor.FILTER_KEY)\n        self._glue = boto3.client('glue')\n        self._extract_iter: Union[None, Iterator] = None\n\n    def extract(self) -> Union[TableMetadata, None]:\n        if not self._extract_iter:\n            self._extract_iter = self._get_extract_iter()\n        try:\n            return next(self._extract_iter)\n        except StopIteration:\n            return None\n\n    def get_scope(self) -> str:\n        return 'extractor.glue'\n\n    def _get_extract_iter(self) -> Iterator[TableMetadata]:\n        \"\"\"\n        It gets all tables and yields TableMetadata\n        :return:\n        \"\"\"\n        for row in self._get_raw_extract_iter():\n            columns, i = [], 0\n\n            for column in row['StorageDescriptor']['Columns'] \\\n                    + row.get('PartitionKeys', []):\n                columns.append(ColumnMetadata(\n                    column['Name'],\n                    column['Comment'] if 'Comment' in column else None,\n                    column['Type'],\n                    i\n                ))\n                i += 1\n\n            yield TableMetadata(\n                'glue',\n                self._cluster,\n                row['DatabaseName'],\n                row['Name'],\n                row.get('Description') or row.get('Parameters', {}).get('comment'),\n                columns,\n                row.get('TableType') == 'VIRTUAL_VIEW',\n            )\n\n    def _get_raw_extract_iter(self) -> Iterator[Dict[str, Any]]:\n        \"\"\"\n        Provides iterator of results row from glue client\n        :return:\n        \"\"\"\n        tables = self._search_tables()\n        return iter(tables)\n\n    def _search_tables(self) -> List[Dict[str, Any]]:\n        tables = []\n        kwargs = {}\n        if self._filters is not None:\n            kwargs['Filters'] = self._filters\n        data = self._glue.search_tables(**kwargs)\n        tables += data['TableList']\n        while 'NextToken' in data:\n            token = data['NextToken']\n            kwargs['NextToken'] = token\n            data = self._glue.search_tables(**kwargs)\n            tables += data['TableList']\n        return tables\n", "path": "databuilder/databuilder/extractor/glue_extractor.py"}], "after_files": [{"content": "# Copyright Contributors to the Amundsen project.\n# SPDX-License-Identifier: Apache-2.0\n\nfrom typing import (\n    Any, Dict, Iterator, List, Union,\n)\n\nimport boto3\nfrom pyhocon import ConfigFactory, ConfigTree\n\nfrom databuilder.extractor.base_extractor import Extractor\nfrom databuilder.models.table_metadata import ColumnMetadata, TableMetadata\n\n\nclass GlueExtractor(Extractor):\n    \"\"\"\n    Extracts tables and columns metadata from AWS Glue metastore\n    \"\"\"\n\n    CLUSTER_KEY = 'cluster'\n    FILTER_KEY = 'filters'\n    MAX_RESULTS_KEY = 'max_results'\n    DEFAULT_CONFIG = ConfigFactory.from_dict({CLUSTER_KEY: 'gold', FILTER_KEY: None, MAX_RESULTS_KEY: 500})\n\n    def init(self, conf: ConfigTree) -> None:\n        conf = conf.with_fallback(GlueExtractor.DEFAULT_CONFIG)\n        self._cluster = conf.get_string(GlueExtractor.CLUSTER_KEY)\n        self._filters = conf.get(GlueExtractor.FILTER_KEY)\n        self._max_results = conf.get(GlueExtractor.MAX_RESULTS_KEY)\n        self._glue = boto3.client('glue')\n        self._extract_iter: Union[None, Iterator] = None\n\n    def extract(self) -> Union[TableMetadata, None]:\n        if not self._extract_iter:\n            self._extract_iter = self._get_extract_iter()\n        try:\n            return next(self._extract_iter)\n        except StopIteration:\n            return None\n\n    def get_scope(self) -> str:\n        return 'extractor.glue'\n\n    def _get_extract_iter(self) -> Iterator[TableMetadata]:\n        \"\"\"\n        It gets all tables and yields TableMetadata\n        :return:\n        \"\"\"\n        for row in self._get_raw_extract_iter():\n            columns, i = [], 0\n\n            for column in row['StorageDescriptor']['Columns'] \\\n                    + row.get('PartitionKeys', []):\n                columns.append(ColumnMetadata(\n                    column['Name'],\n                    column['Comment'] if 'Comment' in column else None,\n                    column['Type'],\n                    i\n                ))\n                i += 1\n\n            yield TableMetadata(\n                'glue',\n                self._cluster,\n                row['DatabaseName'],\n                row['Name'],\n                row.get('Description') or row.get('Parameters', {}).get('comment'),\n                columns,\n                row.get('TableType') == 'VIRTUAL_VIEW',\n            )\n\n    def _get_raw_extract_iter(self) -> Iterator[Dict[str, Any]]:\n        \"\"\"\n        Provides iterator of results row from glue client\n        :return:\n        \"\"\"\n        tables = self._search_tables()\n        return iter(tables)\n\n    def _search_tables(self) -> List[Dict[str, Any]]:\n        tables = []\n        kwargs = {}\n        if self._filters is not None:\n            kwargs['Filters'] = self._filters\n            kwargs['MaxResults'] = self._max_results\n        data = self._glue.search_tables(**kwargs)\n        tables += data['TableList']\n        while 'NextToken' in data:\n            token = data['NextToken']\n            kwargs['NextToken'] = token\n            data = self._glue.search_tables(**kwargs)\n            tables += data['TableList']\n        return tables\n", "path": "databuilder/databuilder/extractor/glue_extractor.py"}]}