AWS Athena调用getQueryExecution返回400错误求助

我用NodeJS AWS SDK调用AWS.Athena.getQueryExecution查询Athena任务状态时，收到以下错误：

{
    "message": "Query has not yet finished. Current state: QUEUED",
    "code": "InvalidRequestException",
    "time": "2023-01-27T16:51:23.355Z",
    "requestId": "c64ee58c-e326-4223-8d80-8323a25e8fe0",
    "statusCode": 400,
    "retryable": false,
    "retryDelay": 89.10527399170958
}

明明能识别任务处于QUEUED状态，但为什么会返回400状态码？我需要获取状态来等待任务完成，希望得到包含状态的非错误响应，循环检查直到任务结束。

我的代码如下：

import sleep from "../sleep.js"
import AWS from 'aws-sdk'

export default async (credentials) => {
    const athenaConn = new AWS.Athena({
            accessKeyId: data.Credentials.AccessKeyId,
            secretAccessKey: data.Credentials.SecretAccessKey,
            sessionToken: data.Credentials.SessionToken
    })

    const startQueryExecutionResponse = await athena.startQueryExecution(params = { QueryString: "select * from database.table" }).promise()

    let status = "UNKNOWN"
    do {
        await sleep(1000)
        status = await getQueryStatus(athenaConn, startQueryExecutionResponse)
    } while (status === "QUEUED" || status === "RUNNING")

    return await athena.getQueryResults(startQueryExecutionResponse).promise()
}

const getQueryStatus = async (athenaConn, startQueryExecutionResponse) => {
    const result = await athenaConn.getQueryExecution(startQueryExecutionResponse).promise()
    return result.QueryExecution.Status.State || "Unknown"
}

问题原因 #

你错误地将startQueryExecution的完整响应对象传给了getQueryExecution，而该API仅接受包含QueryExecutionId的对象作为入参。多余的字段被发送到AWS API后，触发了InvalidRequestException错误，即便API能识别任务当前状态。

修复代码 #

import sleep from "../sleep.js"
import AWS from 'aws-sdk'

export default async (credentials) => {
    // 修正变量名错误：data → credentials
    const athenaConn = new AWS.Athena({
            accessKeyId: credentials.Credentials.AccessKeyId,
            secretAccessKey: credentials.Credentials.SecretAccessKey,
            sessionToken: credentials.Credentials.SessionToken
    })

    const startQueryExecutionResponse = await athenaConn.startQueryExecution({ 
        QueryString: "select * from database.table" 
    }).promise()

    const queryExecutionId = startQueryExecutionResponse.QueryExecutionId;
    let status = "UNKNOWN"
    do {
        await sleep(1000)
        try {
            status = await getQueryStatus(athenaConn, queryExecutionId)
        } catch (err) {
            // 针对状态未完成的错误，直接从错误信息提取状态继续循环
            if (err.code === 'InvalidRequestException' && err.message.includes('Current state:')) {
                status = err.message.split('Current state: ')[1].trim();
            } else {
                // 其他错误直接抛出
                throw err;
            }
        }
    } while (status === "QUEUED" || status === "RUNNING")

    return await athenaConn.getQueryResults({ QueryExecutionId: queryExecutionId }).promise()
}

const getQueryStatus = async (athenaConn, queryExecutionId) => {
    // 仅传入QueryExecutionId参数
    const result = await athenaConn.getQueryExecution({ 
        QueryExecutionId: queryExecutionId 
    }).promise()
    return result.QueryExecution.Status.State || "Unknown"
}

额外说明 #

1. 原代码存在变量名错误：函数参数是credentials，但内部使用了未定义的data，已修正。
2. 部分场景下，即便传入正确参数，Athena API仍可能在任务处于QUEUED/RUNNING时返回400错误，此时可从错误信息中提取状态继续循环，避免流程中断。

内容的提问来源于stack exchange，提问作者marifkind