@Override
    public void configure(Context context) {
        LOGGER.info(context.toString());
        masterHostAndPortList = context.getString(MASTER_HOST_PORT_LIST);
        defaultTopic = context.getString(TOPIC);
        heartbeatPeriod = context.getLong(HEARTBEAT_PERIOD, DEFAULT_HEARTBEAT_PERIOD);
        rpcTimeout = context.getLong(RPC_TIMEOUT, DEFAULT_RPC_TIMEOUT);

        linkMaxAllowedDelayedMsgCount = context.getLong(
                LINK_MAX_ALLOWED_DELAYED_MSG_COUNT,
                DEFAULT_LINK_MAX_ALLOWED_DELAYED_MSG_COUNT);
        sessionWarnDelayedMsgCount = context.getLong(
                SESSION_WARN_DELAYED_MSG_COUNT,
                DEFAULT_SESSION_WARN_DELAYED_MSG_COUNT);
        sessionMaxAllowedDelayedMsgCount = context.getLong(
                SESSION_MAX_ALLOWED_DELAYED_MSG_COUNT,
                DEFAULT_SESSION_MAX_ALLOWED_DELAYED_MSG_COUNT);
        nettyWriteBufferHighWaterMark = context.getLong(
                NETTY_WRITE_BUFFER_HIGH_WATER_MARK,
                DEFAULT_NETTY_WRITE_BUFFER_HIGH_WATER_MARK);

        producerMap = new ConcurrentHashMap<>();

        threadNum = context.getInteger(SINK_THREAD_NUM, DEFAULT_SINK_THREAD_NUM);
        sinkThreadPool = Executors.newFixedThreadPool(threadNum);

        int retryQueueCapacity = context.getInteger(RETRY_QUEUE_CAPACITY, DEFAULT_RETRY_QUEUE_CAPACITY);
        resendQueue = new LinkedBlockingQueue<>(retryQueueCapacity);

        int eventQueueCapacity = context.getInteger(EVENT_QUEUE_CAPACITY, DEFAULT_EVENT_QUEUE_CAPACITY);
        eventQueue = new LinkedBlockingQueue<>(eventQueueCapacity);

        maxRetryTime = context.getInteger(EVENT_MAX_RETRY_TIME, DEFAULT_EVENT_MAX_RETRY_TIME);
        eventOfferTimeout = context.getLong(EVENT_OFFER_TIMEOUT, DEFAULT_EVENT_OFFER_TIMEOUT);

        counter = new TubeSinkCounter(this.getName());

        clientConfig = initTubeConfig();
    }
Destroy all producers and clear up caches.
inlong-tubemq/tubemq-connectors/tubemq-connector-flume/src/main/java/org/apache/flume/sink/tubemq/TubemqSink.java