optimize this codepublic class CallStatusThread implements Runnable private HashMapInteger ListCampaignBO campaignMap; private HashMapInteger String templateMap; private BotServiceConsumer c

Here are some possible optimizations:

1. Use ConcurrentHashMap instead of HashMap for thread-safety in case multiple threads access the campaignMap and templateMap.

2. Use a single loop to iterate through the callOutMessageBOS and campaignBOS lists instead of nested loops, to reduce the number of iterations.

3. Use a lambda expression instead of creating a new Runnable implementation for the EXECUTOR_SERVICE, to reduce code verbosity.

4. Use a try-with-resources statement to automatically close the consumer when it is no longer needed.

5. Use a logger instance specific to the class instead of a static logger, to avoid potential concurrency issues.

Here's the optimized code:

import java.util.; import java.util.concurrent.; import org.apache.kafka.clients.consumer.; import org.apache.log4j.;

public class CallStatusThread implements Runnable {

private Map<Integer, List<CampaignBO>> campaignMap;
private Map<Integer, String> templateMap;
private BotServiceConsumer consumer;
private static final Logger LOGGER = LogManager.getLogger(CallStatusThread.class);
private static final ExecutorService EXECUTOR_SERVICE = Executors.newFixedThreadPool(Constants.UpdateCall.POOL_SIZE);
private Thread thread;

public CallStatusThread() {
    thread = new Thread(this);
    campaignMap = new ConcurrentHashMap<>();
    templateMap = new ConcurrentHashMap<>();
    List<CampaignBO> campaignBOS = new CampaignDAL().getCampaignsConfig();
    Set<Integer> botIdSet = new HashSet<>();
    for (CampaignBO campaignBO : campaignBOS) {
        botIdSet.add(campaignBO.getBotId());
    }
    for (Integer botId : botIdSet) {
        campaignMap.put(botId, new CampaignDAL().getCampaignsConfig(botId));
    }
    List<CallOutMessageBO> callOutMessageBOS = new CallOutMessageDAL().getMessagesBO();
    for (CallOutMessageBO callOutMessageBO : callOutMessageBOS) {
        int botId = callOutMessageBO.getBotId();
        templateMap.computeIfAbsent(botId, k -> callOutMessageBO.getTemplatePostCall());
    }
    consumer = new BotServiceConsumer(Constants.UpdateCall.Kafka.BROKER, Constants.UpdateCall.Kafka.STATUS_TOPIC, Constants.UpdateCall.Kafka.STATUS_GROUP, Constants.UpdateCall.Kafka.MAX_POLL_RECORDS);
}

private void processingCallStatus() throws ParseException {
    try (consumer) {
        consumer.subscribe();
        CallOutInfoDAL connector = new CallOutInfoDAL();
        LOGGER.info("Starting UpdateCallStatusThread ...");
        while (true) {
            ConsumerRecords<String, String> records = consumer.getRecords();
            if (records == null || records.isEmpty()) {
                continue;
            }
            LOGGER.info("Consume " + records.count() + " messages");
            List<CallOutInfoBO> callStatusBOS = new ArrayList<>();
            for (ConsumerRecord<String, String> record : records) {
                try {
                    CallOutInfoBO bo = MappingObjectJson.convertJsonToObject(record.value(), CallOutInfoBO.class);
                    callStatusBOS.add(bo);
                    int botId = bo.getBotId();
                    int maxRetry = campaignMap.get(botId).get(0).getRetryTime();
                    int callTimes = bo.getTimeCallOfDay();
                    String template = templateMap.get(botId);
                    if (template == null || template.isEmpty()) {
                        LOGGER.info("Don't Update with botId = " + botId + ", " + bo);
                    } else {
                        EXECUTOR_SERVICE.execute(() -> {
                            try {
                                boolean isUpdated = update(bo, template, botId);
                                LOGGER.info("Update result : " + isUpdated + ", botId=" + botId + ", " + bo);
                            } catch (Exception e) {
                                LOGGER.error("Update ERROR, botId = " + botId + ", " + bo, e);
                            }
                        });
                    }
                } catch (Exception e) {
                    LOGGER.error("Process message " + record.value() + " error", e);
                }
            }
            CallOutInfoBO[] callStatusArray = callStatusBOS.toArray(new CallOutInfoBO[0]);
            connector.insert(callStatusArray);
        }
    }
}

public boolean update(CallOutInfoBO bo, String template, int botId) throws Exception {
    if (botId == 9) {
        return HttpClientFactory.updateCallStatusToOrder(bo);
    }
    return true;
}

public void start() {
    thread.start();
}

@Override
public void run() {
    try {
        processingCallStatus();
    } catch (ParseException e) {
        LOGGER.error("processingCallStatus error", e);
    }
}

}