<commit_before>package com.ctrip.hermes.producer.monitor;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.unidal.lookup.annotation.Named;
import org.unidal.tuple.Pair;

import com.ctrip.hermes.core.constants.CatConstants;
import com.ctrip.hermes.core.message.ProducerMessage;
import com.ctrip.hermes.core.transport.command.SendMessageCommand;
import com.ctrip.hermes.core.transport.command.SendMessageResultCommand;
import com.ctrip.hermes.producer.status.ProducerStatusMonitor;
import com.dianping.cat.Cat;
import com.dianping.cat.message.Transaction;
import com.dianping.cat.message.internal.DefaultTransaction;
import com.dianping.cat.message.spi.MessageTree;
import com.google.common.util.concurrent.SettableFuture;

/**
 * @author <NAME>(<EMAIL>)
 *
 */
@Named(type = SendMessageResultMonitor.class)
public class DefaultSendMessageResultMonitor implements SendMessageResultMonitor {
	private static final Logger log = LoggerFactory.getLogger(DefaultSendMessageResultMonitor.class);

	private Map<Long, Pair<SendMessageCommand, SettableFuture<Void>>> m_cmds = new ConcurrentHashMap<>();

	@Override
	public Future<Void> monitor(SendMessageCommand cmd) {
		SettableFuture<Void> future = SettableFuture.create();
		m_cmds.put(cmd.getHeader().getCorrelationId(), new Pair<SendMessageCommand, SettableFuture<Void>>(cmd, future));
		return future;
	}

	@Override
	public void resultReceived(SendMessageResultCommand result) {
		if (result != null) {
			Pair<SendMessageCommand, SettableFuture<Void>> pair = m_cmds.remove(result.getHeader().getCorrelationId());
			if (pair != null) {
				try {
					SendMessageCommand sendMessageCommand = pair.getKey();
					SettableFuture<Void> future = pair.getValue();

					ProducerStatusMonitor.INSTANCE.brokerResultReceived(sendMessageCommand.getTopic(),
					      sendMessageCommand.getPartition(), sendMessageCommand.getMessageCount());

					future.set(null);
					sendMessageCommand.onResultReceived(result);
					tracking(sendMessageCommand, true);
				} catch (Exception e) {
					log.warn("Exception occurred while calling resultReceived", e);
				}

			}
		}
	}

	private void tracking(SendMessageCommand sendMessageCommand, boolean success) {
		String status = success ? Transaction.SUCCESS : "Timeout";

		for (List<ProducerMessage<?>> msgs : sendMessageCommand.getProducerMessages()) {
			for (ProducerMessage<?> msg : msgs) {
				Transaction t = Cat.newTransaction("Message.Produce.Acked", msg.getTopic());
				MessageTree tree = Cat.getManager().getThreadLocalMessageTree();

				String msgId = msg.getDurableSysProperty(CatConstants.SERVER_MESSAGE_ID);
				String parentMsgId = msg.getDurableSysProperty(CatConstants.CURRENT_MESSAGE_ID);
				String rootMsgId = msg.getDurableSysProperty(CatConstants.ROOT_MESSAGE_ID);

				tree.setMessageId(msgId);
				tree.setParentMessageId(parentMsgId);
				tree.setRootMessageId(rootMsgId);

				Transaction elapseT = Cat.newTransaction("Message.Produce.Elapse", msg.getTopic());
				if (elapseT instanceof DefaultTransaction) {
					((DefaultTransaction) elapseT).setDurationStart(msg.getBornTimeNano());
					elapseT.addData("command.message.count", sendMessageCommand.getMessageCount());
				}
				elapseT.setStatus(status);
				elapseT.complete();

				t.setStatus(status);
				t.complete();
			}
		}
	}

	@Override
	public void cancel(SendMessageCommand cmd) {
		m_cmds.remove(cmd.getHeader().getCorrelationId());
	}

}
<commit_msg>add lock in DefaultSendMessageResultMonitor<commit_after>package com.ctrip.hermes.producer.monitor;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.unidal.lookup.annotation.Named;
import org.unidal.tuple.Pair;

import com.ctrip.hermes.core.constants.CatConstants;
import com.ctrip.hermes.core.message.ProducerMessage;
import com.ctrip.hermes.core.transport.command.SendMessageCommand;
import com.ctrip.hermes.core.transport.command.SendMessageResultCommand;
import com.ctrip.hermes.producer.status.ProducerStatusMonitor;
import com.dianping.cat.Cat;
import com.dianping.cat.message.Transaction;
import com.dianping.cat.message.internal.DefaultTransaction;
import com.dianping.cat.message.spi.MessageTree;
import com.google.common.util.concurrent.SettableFuture;

/**
 * @author <NAME>(<EMAIL>)
 *
 */
@Named(type = SendMessageResultMonitor.class)
public class DefaultSendMessageResultMonitor implements SendMessageResultMonitor {
	private static final Logger log = LoggerFactory.getLogger(DefaultSendMessageResultMonitor.class);

	private Map<Long, Pair<SendMessageCommand, SettableFuture<Void>>> m_cmds = new ConcurrentHashMap<>();

	private ReentrantLock m_lock = new ReentrantLock();

	@Override
	public Future<Void> monitor(SendMessageCommand cmd) {
		m_lock.lock();
		try {
			SettableFuture<Void> future = SettableFuture.create();
			m_cmds.put(cmd.getHeader().getCorrelationId(), new Pair<SendMessageCommand, SettableFuture<Void>>(cmd, future));
			return future;
		} finally {
			m_lock.unlock();
		}
	}

	@Override
	public void resultReceived(SendMessageResultCommand result) {
		if (result != null) {
			Pair<SendMessageCommand, SettableFuture<Void>> pair = null;
			m_lock.lock();
			try {
				pair = m_cmds.remove(result.getHeader().getCorrelationId());
			} finally {
				m_lock.unlock();
			}
			if (pair != null) {
				try {
					SendMessageCommand sendMessageCommand = pair.getKey();
					SettableFuture<Void> future = pair.getValue();

					ProducerStatusMonitor.INSTANCE.brokerResultReceived(sendMessageCommand.getTopic(),
					      sendMessageCommand.getPartition(), sendMessageCommand.getMessageCount());

					future.set(null);
					sendMessageCommand.onResultReceived(result);
					tracking(sendMessageCommand, true);
				} catch (Exception e) {
					log.warn("Exception occurred while calling resultReceived", e);
				}

			}
		}
	}

	private void tracking(SendMessageCommand sendMessageCommand, boolean success) {
		String status = success ? Transaction.SUCCESS : "Timeout";

		for (List<ProducerMessage<?>> msgs : sendMessageCommand.getProducerMessages()) {
			for (ProducerMessage<?> msg : msgs) {
				Transaction t = Cat.newTransaction("Message.Produce.Acked", msg.getTopic());
				MessageTree tree = Cat.getManager().getThreadLocalMessageTree();

				String msgId = msg.getDurableSysProperty(CatConstants.SERVER_MESSAGE_ID);
				String parentMsgId = msg.getDurableSysProperty(CatConstants.CURRENT_MESSAGE_ID);
				String rootMsgId = msg.getDurableSysProperty(CatConstants.ROOT_MESSAGE_ID);

				tree.setMessageId(msgId);
				tree.setParentMessageId(parentMsgId);
				tree.setRootMessageId(rootMsgId);

				Transaction elapseT = Cat.newTransaction("Message.Produce.Elapse", msg.getTopic());
				if (elapseT instanceof DefaultTransaction) {
					((DefaultTransaction) elapseT).setDurationStart(msg.getBornTimeNano());
					elapseT.addData("command.message.count", sendMessageCommand.getMessageCount());
				}
				elapseT.setStatus(status);
				elapseT.complete();

				t.setStatus(status);
				t.complete();
			}
		}
	}

	@Override
	public void cancel(SendMessageCommand cmd) {
		m_lock.lock();
		try {
			m_cmds.remove(cmd.getHeader().getCorrelationId());
		} finally {
			m_lock.unlock();
		}
	}

}

{'fraction_non_alphanumeric': 0.11604050093258726, 'fraction_numerical': 0.0005329070077271516, 'mean_word_length': 5.3137089991589574, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4675260', 'n_tokens_mistral': 2520, 'n_tokens_neox': 2257, 'n_words': 370}