private boolean triggerFlushAndAddMsg(boolean needAdd, boolean isTimeTrigger,
            int partitionId, int keyCode,
            long receivedTime, ByteBuffer indexEntry,
            int dataLength, ByteBuffer dataEntry,
            AppendResult appendResult) throws IOException {
        long startTime;
        writeCacheMutex.writeLock().lock();
        try {
            if (!isFlushOngoing.get() && hasFlushBeenTriggered.compareAndSet(false, true)) {
                this.executor.execute(new Runnable() {

                    @Override
                    public void run() {
                        try {
                            final StringBuilder strBuffer = new StringBuilder(512);
                            flush(strBuffer);
                        } catch (Throwable e) {
                            logger.error("[Data Store] Error during flush", e);
                        } finally {
                            if (isTimeTrigger) {
                                msgStoreStatsHolder.addCacheTimeoutFlush();
                            }
                        }
                    }
                });
            } else {
                msgStoreStatsHolder.addCachePending();
            }
            startTime = System.currentTimeMillis();
            while (hasFlushBeenTriggered.get()) {
                flushWriteCacheCondition.awaitNanos(FLUSH_CONDITION_WAIT_DLT_NS);
                if (System.currentTimeMillis() - startTime > 2000) {
                    logger.warn(new StringBuilder(512)
                            .append("[Data Store] StoreKey=").append(storeKey)
                            .append(" Wait Cache flush write too long! wait time is ")
                            .append(System.currentTimeMillis() - startTime).toString());
                    break;
                }
            }
            if (needAdd) {
                return msgMemStore.appendMsg(msgStoreStatsHolder, partitionId, keyCode,
                        receivedTime, indexEntry, dataLength, dataEntry, appendResult);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IOException(new StringBuilder(512)
                    .append("[Data Store] StoreKey=").append(storeKey)
                    .append(" Interrupted when triggerFlushAndAddMsg process for storekey ")
                    .append(storeKey).toString());
        } finally {
            writeCacheMutex.writeLock().unlock();
        }
        return false;
    }
Append message and trigger flush operation.

@param needAdd           whether to add a message
@param isTimeTrigger     whether is timer trigger
@param partitionId       the partitionId for reading messages
@param keyCode           the filter item hash code
@param receivedTime      the received time of message
@param indexEntry        the stored index entry
@param dataLength        the stored data entry length
@param dataEntry         the stored data entry
@param appendResult      the append result

@return                  the append result
@throws IOException      the exception during processing
inlong-tubemq/tubemq-server/src/main/java/org/apache/inlong/tubemq/server/broker/msgstore/MessageStore.java