public void run()

in adapter/runtime/src/main/java/org/apache/rocketmq/eventbridge/adapter/runtime/boot/EventTargetTrigger.java [59:86]

27 lines of code
6 McCabe index (conditional complexity)


    public void run() {
        while (!stopped) {
            Map<String, List<ConnectRecord>> targetRecordMap = circulatorContext.takeTargetRecords(batchSize);
            if (MapUtils.isEmpty(targetRecordMap)) {
                logger.trace("current target pusher is empty");
                this.waitForRunning(1000);
                continue;
            }
            if (logger.isDebugEnabled()) {
                logger.debug("start push content by pusher - {}", JSON.toJSONString(targetRecordMap));
            }

            for(String runnerName: targetRecordMap.keySet()){
                ExecutorService executorService = circulatorContext.getExecutorService(runnerName);
                executorService.execute(() -> {
                    SinkTask sinkTask = circulatorContext.getPusherTaskMap().get(runnerName);
                    List<ConnectRecord> triggerRecords = targetRecordMap.get(runnerName);
                    try {
                        sinkTask.put(triggerRecords);
                        offsetManager.commit(triggerRecords);
                    } catch (Exception exception) {
                        logger.error(getServiceName() + " push target exception, stackTrace-", exception);
                        triggerRecords.forEach(triggerRecord -> errorHandler.handle(triggerRecord, exception));
                    }
                });
            }
        }
    }