### Question:
HelixTaskExecutor implements MessageListener, TaskExecutor { @Override public void registerMessageHandlerFactory(String type, MessageHandlerFactory factory) { registerMessageHandlerFactory(type, factory, DEFAULT_PARALLEL_TASKS); } HelixTaskExecutor();  HelixTaskExecutor(ParticipantStatusMonitor participantStatusMonitor);  HelixTaskExecutor(ParticipantStatusMonitor participantStatusMonitor,
      MessageQueueMonitor messageQueueMonitor); @Override void registerMessageHandlerFactory(String type, MessageHandlerFactory factory); @Override void registerMessageHandlerFactory(String type, MessageHandlerFactory factory,
      int threadpoolSize); void setController(GenericHelixController controller); ParticipantStatusMonitor getParticipantMonitor(); @Override List<Future<HelixTaskResult>> invokeAllTasks(List<MessageTask> tasks, long timeout,
      TimeUnit unit); @Override boolean cancelTimeoutTask(MessageTask task); @Override boolean scheduleTask(MessageTask task); @Override boolean cancelTask(MessageTask task); @Override void finishTask(MessageTask task); @Override @PreFetch(enabled = false) void onMessage(String instanceName, List<Message> messages,
      NotificationContext changeContext); MessageHandler createMessageHandler(Message message, NotificationContext changeContext); @Override void shutdown(); static final int DEFAULT_PARALLEL_TASKS; static final String MAX_THREADS; }### Answer:
@Test public void testNoWriteReadStateForRemovedMessage() throws NoSuchMethodException, InvocationTargetException, IllegalAccessException { HelixTaskExecutor executor = new HelixTaskExecutor(); HelixManager manager = new MockClusterManager(); TestMessageHandlerFactory factory = new TestMessageHandlerFactory(); for (String type : factory.getMessageTypes()) { executor.registerMessageHandlerFactory(type, factory); } HelixDataAccessor accessor = manager.getHelixDataAccessor(); PropertyKey.Builder keyBuilder = accessor.keyBuilder(); String instanceName = "someInstance"; List<String> messageIds = new ArrayList<>(); List<Message> messages = new ArrayList<>(); int nMsgs1 = 5; for (int i = 0; i < nMsgs1; i++) { Message msg = new Message(factory.getMessageTypes().get(0), UUID.randomUUID().toString()); msg.setTgtSessionId(manager.getSessionId()); msg.setTgtName("Localhost_1123"); msg.setSrcName("127.101.1.23_2234"); msg.setCorrelationId(UUID.randomUUID().toString()); accessor.setProperty(keyBuilder.message(instanceName, msg.getId()), msg); messageIds.add(msg.getId()); messages.add(msg); } Method updateMessageState = HelixTaskExecutor.class .getDeclaredMethod("updateMessageState", List.class, HelixDataAccessor.class, String.class); updateMessageState.setAccessible(true); updateMessageState.invoke(executor, messages, accessor, instanceName); Assert.assertEquals(accessor.getChildNames(keyBuilder.messages(instanceName)).size(), nMsgs1); accessor.removeProperty(keyBuilder.message(instanceName, messageIds.get(0))); System.out.println(accessor.getChildNames(keyBuilder.messages(instanceName)).size()); updateMessageState.invoke(executor, messages, accessor, instanceName); Assert .assertEquals(accessor.getChildNames(keyBuilder.messages(instanceName)).size(), nMsgs1 - 1); }