### Question:
CloudwatchMetadataHandler extends MetadataHandler { @Override public ListTablesResponse doListTables(BlockAllocator blockAllocator, ListTablesRequest listTablesRequest) throws TimeoutException { String logGroupName = tableResolver.validateSchema(listTablesRequest.getSchemaName()); DescribeLogStreamsRequest request = new DescribeLogStreamsRequest(logGroupName); DescribeLogStreamsResult result; List<TableName> tables = new ArrayList<>(); do { if (tables.size() > MAX_RESULTS) { throw new RuntimeException("Too many log streams, exceeded max metadata results for table count."); } result = invoker.invoke(() -> awsLogs.describeLogStreams(request)); result.getLogStreams().forEach(next -> tables.add(toTableName(listTablesRequest, next))); request.setNextToken(result.getNextToken()); logger.info("doListTables: Listing log streams {} {}", result.getNextToken(), tables.size()); } while (result.getNextToken() != null); tables.add(new TableName(listTablesRequest.getSchemaName(), ALL_LOG_STREAMS_TABLE)); return new ListTablesResponse(listTablesRequest.getCatalogName(), tables); } CloudwatchMetadataHandler(); @VisibleForTesting protected  CloudwatchMetadataHandler(AWSLogs awsLogs,
            EncryptionKeyFactory keyFactory,
            AWSSecretsManager secretsManager,
            AmazonAthena athena,
            String spillBucket,
            String spillPrefix); @Override ListSchemasResponse doListSchemaNames(BlockAllocator blockAllocator, ListSchemasRequest listSchemasRequest); @Override ListTablesResponse doListTables(BlockAllocator blockAllocator, ListTablesRequest listTablesRequest); @Override GetTableResponse doGetTable(BlockAllocator blockAllocator, GetTableRequest getTableRequest); @Override void enhancePartitionSchema(SchemaBuilder partitionSchemaBuilder, GetTableLayoutRequest request); @Override void getPartitions(BlockWriter blockWriter, GetTableLayoutRequest request, QueryStatusChecker queryStatusChecker); @Override GetSplitsResponse doGetSplits(BlockAllocator allocator, GetSplitsRequest request);  }### Answer:
@Test public void doListTables() throws TimeoutException { logger.info("doListTables - enter"); when(mockAwsLogs.describeLogStreams(any(DescribeLogStreamsRequest.class))).thenAnswer((InvocationOnMock invocationOnMock) -> { DescribeLogStreamsRequest request = (DescribeLogStreamsRequest) invocationOnMock.getArguments()[0]; DescribeLogStreamsResult result = new DescribeLogStreamsResult(); Integer nextToken; if (request.getNextToken() == null) { nextToken = 1; } else if (Integer.valueOf(request.getNextToken()) < 3) { nextToken = Integer.valueOf(request.getNextToken()) + 1; } else { nextToken = null; } List<LogStream> logStreams = new ArrayList<>(); if (request.getNextToken() == null || Integer.valueOf(request.getNextToken()) < 3) { for (int i = 0; i < 10; i++) { LogStream nextLogStream = new LogStream(); nextLogStream.setLogStreamName("table-" + String.valueOf(i)); logStreams.add(nextLogStream); } } result.withLogStreams(logStreams); if (nextToken != null) { result.setNextToken(String.valueOf(nextToken)); } return result; }); ListTablesRequest req = new ListTablesRequest(identity, "queryId", "default", "schema-1"); ListTablesResponse res = handler.doListTables(allocator, req); logger.info("doListTables - {}", res.getTables()); assertTrue(res.getTables().contains(new TableName("schema-1", "all_log_streams"))); assertTrue(res.getTables().size() == 31); verify(mockAwsLogs, times(4)).describeLogStreams(any(DescribeLogStreamsRequest.class)); verify(mockAwsLogs, times(1)).describeLogGroups(any(DescribeLogGroupsRequest.class)); verifyNoMoreInteractions(mockAwsLogs); logger.info("doListTables - exit"); }