使用C++ Arrow Flight，报invalid bearer token的错误

[Semon120]

我使用C++ Arrow Flight连接Doris Arrow flight端口，执行SHOW语句没问题，但执行select语句时，会报 invalid bearer token的错误。我确认了传入的token是正确的26为字符。但日志显示出错的token并不是我传入的token。下面是日志：
2024-04-19 13:21:22,052 INFO (flight-server-default-executor-1|272) [FlightTokenManagerImpl.createToken():99] Created flight token for user: root, token: 4r2358sd5shv01ddcuqf4u97vs
2024-04-19 13:21:22,057 INFO (flight-server-default-executor-0|271) [FlightTokenManagerImpl.validateToken():117] Validated bearer token for user: root
2024-04-19 13:21:22,058 INFO (flight-server-default-executor-0|271) [FlightTokenManagerImpl.validateToken():117] Validated bearer token for user: root
2024-04-19 13:21:22,076 INFO (flight-server-default-executor-0|271) [Coordinator.execInternal():702] dispatch result sink of query cacb5da31aeb4100-8fd45d47f5ab04ae to TNetworkAddress(hostname:192.168.0.232, port:34020)
2024-04-19 13:21:22,101 INFO (flight-server-default-executor-0|271) [StmtExecutor.handleQueryStmt():1797] Query cacb5da31aeb4100-8fd45d47f5ab04ae finished
2024-04-19 13:21:22,114 INFO (flight-server-default-executor-0|271) [FlightTokenManagerImpl.validateToken():117] Validated bearer token for user: root
2024-04-19 13:21:22,115 WARN (flight-server-default-executor-0|271) [FlightSessionsWithTokenManager.getConnectContext():54] get ConnectContext failed, invalid bearer token: cacb5da31aeb4100-8fd45d47f5ab04b9, try reconnect, bearer token may not be created, or may have been evict, search for this token in fe.log to see the evict reason. currently in fe.conf, arrow_flight_token_cache_size=512, arrow_flight_token_alive_time=1000
java.lang.IllegalArgumentException: invalid bearer token: cacb5da31aeb4100-8fd45d47f5ab04b9, try reconnect, bearer token may not be created, or may have been evict, search for this token in fe.log to see the evict reason. currently in fe.conf, arrow_flight_token_cache_size=512, arrow_flight_token_alive_time=1000
at org.apache.doris.service.arrowflight.tokens.FlightTokenManagerImpl.validateToken(FlightTokenManagerImpl.java:107) ~[doris-fe.jar:1.2-SNAPSHOT]
at org.apache.doris.service.arrowflight.sessions.FlightSessionsWithTokenManager.createConnectContext(FlightSessionsWithTokenManager.java:61) ~[doris-fe.jar:1.2-SNAPSHOT]
at org.apache.doris.service.arrowflight.sessions.FlightSessionsWithTokenManager.getConnectContext(FlightSessionsWithTokenManager.java:49) ~[doris-fe.jar:1.2-SNAPSHOT]
at org.apache.doris.service.arrowflight.DorisFlightSqlProducer.getStreamStatementResult(DorisFlightSqlProducer.java:131) ~[doris-fe.jar:1.2-SNAPSHOT]
at org.apache.doris.service.arrowflight.DorisFlightSqlProducer.getStreamStatement(DorisFlightSqlProducer.java:162) ~[doris-fe.jar:1.2-SNAPSHOT]
at org.apache.arrow.flight.sql.FlightSqlProducer.getStream(FlightSqlProducer.java:229) ~[flight-sql-15.0.2.jar:15.0.2]
at org.apache.arrow.flight.FlightService.doGetCustom(FlightService.java:111) ~[flight-core-15.0.2.jar:15.0.2]
at org.apache.arrow.flight.FlightBindingService$DoGetMethod.invoke(FlightBindingService.java:144) ~[flight-core-15.0.2.jar:15.0.2]
at org.apache.arrow.flight.FlightBindingService$DoGetMethod.invoke(FlightBindingService.java:134) ~[flight-core-15.0.2.jar:15.0.2]
at io.grpc.stub.ServerCalls$UnaryServerCallHandler$UnaryServerCallListener.onHalfClose(ServerCalls.java:182) ~[grpc-stub-1.60.1.jar:1.60.1]
at io.grpc.PartialForwardingServerCallListener.onHalfClose(PartialForwardingServerCallListener.java:35) ~[grpc-api-1.60.1.jar:1.60.1]
at io.grpc.ForwardingServerCallListener.onHalfClose(ForwardingServerCallListener.java:23) ~[grpc-api-1.60.1.jar:1.60.1]
at io.grpc.ForwardingServerCallListener$SimpleForwardingServerCallListener.onHalfClose(ForwardingServerCallListener.java:40) ~[grpc-api-1.60.1.jar:1.60.1]
at io.grpc.Contexts$ContextualizedServerCallListener.onHalfClose(Contexts.java:86) ~[grpc-api-1.60.1.jar:1.60.1]
at io.grpc.PartialForwardingServerCallListener.onHalfClose(PartialForwardingServerCallListener.java:35) ~[grpc-api-1.60.1.jar:1.60.1]
at io.grpc.ForwardingServerCallListener.onHalfClose(ForwardingServerCallListener.java:23) ~[grpc-api-1.60.1.jar:1.60.1]
at io.grpc.ForwardingServerCallListener$SimpleForwardingServerCallListener.onHalfClose(ForwardingServerCallListener.java:40) ~[grpc-api-1.60.1.jar:1.60.1]
at io.grpc.Contexts$ContextualizedServerCallListener.onHalfClose(Contexts.java:86) ~[grpc-api-1.60.1.jar:1.60.1]
at io.grpc.internal.ServerCallImpl$ServerStreamListenerImpl.halfClosed(ServerCallImpl.java:351) ~[grpc-core-1.60.1.jar:1.60.1]
at io.grpc.internal.ServerImpl$JumpToApplicationThreadServerStreamListener$1HalfClosed.runInContext(ServerImpl.java:861) ~[grpc-core-1.60.1.jar:1.60.1]
at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37) ~[grpc-core-1.60.1.jar:1.60.1]
at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:133) ~[grpc-core-1.60.1.jar:1.60.1]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[?:?]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[?:?]
at java.lang.Thread.run(Thread.java:833) ~[?:?]

[yagagagaga]

Can you show me the demo?

[Semon120]

ARROW_ASSIGN_OR_RAISE(auto pre ,sql_client->Prepare(call_options1, sql));

    //ARROW_ASSIGN_OR_RAISE(auto info, sql_client->Execute(call_options1, "select * from sfinx.sample_base_data_fix where date='2005-01-04'"));
    //ARROW_ASSIGN_OR_RAISE(auto info, sql_client->Execute(call_options1, "show backends"));
    ARROW_ASSIGN_OR_RAISE(auto info,pre->Execute(call_options1))
    const auto endpoints = info->endpoints();
    bool first = false;
    for (auto i = 0; i < endpoints.size(); i++) {
        auto endpoint = endpoints[i];
        std::cout<<"location size:"<<endpoint.locations.size()<<std::endl;
        if(endpoint.locations.size() > 0) {
            auto loc = endpoint.locations.at(0);
            
            ARROW_ASSIGN_OR_RAISE(auto beClient, FlightClient::Connect(loc,clientOpts));
            auto &ticket = endpoints[i].ticket;
            MyLnxTimer timer1;
            timer1.Start();
            // 之前使用了fe的连接sql_client去获取结果，报错了
            // ARROW_ASSIGN_OR_RAISE(auto stream, sql_client->DoGet(call_options1, ticket));
            ARROW_ASSIGN_OR_RAISE(auto stream, beClient->DoGet(call_options1, ticket));

            auto schema = stream->GetSchema();
            ARROW_RETURN_NOT_OK(schema);

            std::cout << "Schema:" << schema->get()->ToString() << std::endl;
            ARROW_ASSIGN_OR_RAISE(auto table,stream->ToTable());
            
            // std::cout<<table->ToString()<<std::endl;
            timer1.End();
            std::cout<<"endpoint "<<i<<" "<<timer1.toString("cost")<<std::endl;
        }
    }

[Semon120]

目前 问题已经解决了，原因是：获取结果数据流应该去BE节点获取，之前拿到tick再去FE获取结果就会爆token错误。

[yagagagaga]

Hi, @Semon120 if you are interested, you can contribute the code to https://github.com/apache/doris/tree/master/samples