@Override
    public void run() {
        boolean isNeedFlush = false;
        boolean sqlShowEnabled = ProxyContext.getInstance().getContextManager().getMetaDataContexts().getMetaData().getProps().getValue(ConfigurationPropertyKey.SQL_SHOW);
        try {
            if (sqlShowEnabled) {
                fillLogMDC();
            }
            isNeedFlush = executeCommand(context,
                    databaseProtocolFrontendEngine.getCodecEngine().createPacketPayload((ByteBuf) message, context.channel().attr(CommonConstants.CHARSET_ATTRIBUTE_KEY).get()));
            // CHECKSTYLE:OFF
        } catch (final Exception ex) {
            // CHECKSTYLE:ON
            processException(ex);
            // CHECKSTYLE:OFF
        } catch (final Error error) {
            // CHECKSTYLE:ON
            processException(new RuntimeException(error));
        } finally {
            connectionSession.clearQueryContext();
            Collection<SQLException> exceptions = Collections.emptyList();
            try {
                connectionSession.getDatabaseConnectionManager().closeExecutionResources();
            } catch (final BackendConnectionException ex) {
                exceptions = ex.getExceptions().stream().filter(SQLException.class::isInstance).map(SQLException.class::cast).collect(Collectors.toList());
            }
            if (isNeedFlush) {
                context.flush();
            }
            processClosedExceptions(exceptions);
            context.pipeline().fireUserEventTriggered(new WriteCompleteEvent());
            if (sqlShowEnabled) {
                clearLogMDC();
            }
            if (message instanceof CompositeByteBuf) {
                releaseCompositeByteBuf((CompositeByteBuf) message);
            }
            ((ByteBuf) message).release();
        }
    }
@Test
    void assertRunNeedFlushByTrue() throws SQLException, BackendConnectionException {
        when(queryCommandExecutor.execute()).thenReturn(Collections.singleton(databasePacket));
        when(engine.getCommandExecuteEngine().getCommandPacket(payload, commandPacketType, connectionSession)).thenReturn(commandPacket);
        when(engine.getCommandExecuteEngine().getCommandExecutor(commandPacketType, commandPacket, connectionSession)).thenReturn(queryCommandExecutor);
        when(engine.getCommandExecuteEngine().getCommandPacketType(payload)).thenReturn(commandPacketType);
        when(engine.getCodecEngine().createPacketPayload(message, StandardCharsets.UTF_8)).thenReturn(payload);
        CommandExecutorTask actual = new CommandExecutorTask(engine, connectionSession, handlerContext, message);
        actual.run();
        verify(handlerContext).flush();
        verify(engine.getCommandExecuteEngine()).writeQueryData(handlerContext, databaseConnectionManager, queryCommandExecutor, 1);
        verify(queryCommandExecutor).close();
        verify(databaseConnectionManager).closeExecutionResources();
    }