@Override
    public void consume0(@NonNull final List<ShenyuRequestLog> logs) {
        logs.forEach(log -> {
            String logTopic = Optional.ofNullable(LoggingKafkaPluginDataHandler.getSelectApiConfigMap().get(log.getSelectorId()))
                    .map(apiConfig -> StringUtils.defaultIfBlank(apiConfig.getTopic(), topic)
                    ).orElse(topic);
            try {
                LOG.info("logTopic:{}, log:{}", logTopic, log);
                producer.send(toProducerRecord(logTopic, log), (metadata, exception) -> {
                    LOG.info("kafka push logs metadata:{}", GsonUtils.getInstance().toJson(metadata));
                    if (Objects.nonNull(exception)) {
                        LOG.error("kafka push logs error", exception);
                    }
                });
                producer.flush();
            } catch (Exception e) {
                LOG.error("kafka push logs error", e);
            }
        });
    }
store logs.

@param logs list of log
shenyu-plugin/shenyu-plugin-logging/shenyu-plugin-logging-kafka/src/main/java/org/apache/shenyu/plugin/logging/kafka/client/KafkaLogCollectClient.java