/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */
package org.apache.geode.internal.cache.locks;

import static org.apache.geode.cache.Region.SEPARATOR;
import static org.apache.geode.distributed.ConfigurationProperties.LOG_LEVEL;
import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.fail;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;

import org.junit.Test;
import org.junit.experimental.categories.Category;

import org.apache.geode.cache.CommitConflictException;
import org.apache.geode.distributed.DistributedLockService;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.DistributionMessage;
import org.apache.geode.distributed.internal.DistributionMessageObserver;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.distributed.internal.ReplyProcessor21;
import org.apache.geode.distributed.internal.locks.DLockRecoverGrantorProcessor;
import org.apache.geode.distributed.internal.locks.DLockRecoverGrantorProcessor.DLockRecoverGrantorMessage;
import org.apache.geode.distributed.internal.locks.DLockService;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.cache.TXRegionLockRequestImpl;
import org.apache.geode.test.dunit.Host;
import org.apache.geode.test.dunit.Invoke;
import org.apache.geode.test.dunit.LogWriterUtils;
import org.apache.geode.test.dunit.SerializableRunnable;
import org.apache.geode.test.dunit.internal.JUnit4DistributedTestCase;
import org.apache.geode.test.junit.categories.DLockTest;

/**
 * This class tests distributed ownership via the DistributedLockService api.
 */
@Category({DLockTest.class})
public class TXLockServiceDUnitTest extends JUnit4DistributedTestCase {

  private static InternalDistributedSystem system;

  protected static boolean testTXRecoverGrantor_replyCode_PASS = false;
  protected static boolean testTXRecoverGrantor_heldLocks_PASS = false;

  private InternalDistributedMember lockGrantor;

  public TXLockServiceDUnitTest() {
    super();
  }

  // -------------------------------------------------------------------------
  // Lifecycle methods
  // -------------------------------------------------------------------------

  /**
   * Returns a previously created (or new, if this is the first time this method is called in this
   * VM) distributed system which is somewhat configurable via hydra test parameters.
   */
  @Override
  public final void postSetUp() throws Exception {
    Invoke.invokeInEveryVM("connectDistributedSystem", () -> connectDistributedSystem());
    connectDistributedSystem();
  }


  @Override
  public final void preTearDown() throws Exception {
    // invokeInEveryVM(TXLockServiceDUnitTest.class,
    // "remoteDumpAllDLockServices");

    Invoke.invokeInEveryVM(TXLockServiceDUnitTest.class, "destroyServices");

    destroyServices();

    // // Disconnects the DistributedSystem in every VM - since
    // // each test randomly chooses whether shared memory is used
    // disconnectAllFromDS();

    this.lockGrantor = null;

    testTXRecoverGrantor_replyCode_PASS = false;
    testTXRecoverGrantor_heldLocks_PASS = false;
  }

  // -------------------------------------------------------------------------
  // Test methods
  // -------------------------------------------------------------------------

  @Test
  public void testGetAndDestroy() {
    forEachVMInvoke("checkGetAndDestroy", new Object[] {});
    /*
     * invokeInEveryVM(TXLockServiceDUnitTest.class, "destroyServices");
     * forEachVMInvoke("checkGetAndDestroy", new Object[] {});
     */
  }

  @Test
  public void testGetAndDestroyAgain() {
    testGetAndDestroy();
  }

  @Test
  public void testTXRecoverGrantorMessageProcessor() throws Exception {
    TXLockService.createDTLS(system);
    checkDLockRecoverGrantorMessageProcessor();

    /*
     * call TXRecoverGrantorMessageProcessor.process directly to make sure that correct behavior
     * occurs
     */

    // get txLock and hold it
    final Set participants = Collections.EMPTY_SET;
    final List regionLockReqs = new ArrayList();

    Map<Object, Boolean> keymap = new HashMap<Object, Boolean>();
    keymap.put("KEY-1", true);
    keymap.put("KEY-2", true);
    keymap.put("KEY-3", true);
    keymap.put("KEY-4", true);
    regionLockReqs
        .add(new TXRegionLockRequestImpl(SEPARATOR + "testTXRecoverGrantorMessageProcessor",
            keymap));

    TXLockService dtls = TXLockService.getDTLS();
    TXLockId txLockId = dtls.txLock(regionLockReqs, participants);

    // async call TXRecoverGrantorMessageProcessor.process
    final DLockService dlock = ((TXLockServiceImpl) dtls).getInternalDistributedLockService();
    final TestDLockRecoverGrantorProcessor testProc =
        new TestDLockRecoverGrantorProcessor(dlock.getDistributionManager(), Collections.EMPTY_SET);
    assertEquals("No valid processorId", true, testProc.getProcessorId() > -1);

    final DLockRecoverGrantorProcessor.DLockRecoverGrantorMessage msg =
        new DLockRecoverGrantorProcessor.DLockRecoverGrantorMessage();
    msg.setServiceName(dlock.getName());
    msg.setProcessorId(testProc.getProcessorId());
    msg.setSender(dlock.getDistributionManager().getId());

    Thread thread = new Thread(() -> {
      TXRecoverGrantorMessageProcessor proc =
          (TXRecoverGrantorMessageProcessor) dlock.getDLockRecoverGrantorMessageProcessor();
      proc.processDLockRecoverGrantorMessage(dlock.getDistributionManager(), msg);
    });
    thread.setName("TXLockServiceDUnitTest thread");
    thread.setDaemon(true);
    thread.start();

    await("waiting for recovery message to block").until(() -> {
      return ((TXLockServiceImpl) dtls).isRecovering();
    });

    dtls.release(txLockId);

    // check results to verify no locks were provided in the reply
    await("waiting for thread to exit").until(() -> {
      return !thread.isAlive();
    });

    assertFalse(((TXLockServiceImpl) dtls).isRecovering());

    assertEquals("testTXRecoverGrantor_replyCode_PASS is false", true,
        testTXRecoverGrantor_replyCode_PASS);
    assertEquals("testTXRecoverGrantor_heldLocks_PASS is false", true,
        testTXRecoverGrantor_heldLocks_PASS);
  }


  @Test
  public void testTXGrantorMigration() throws Exception {
    // first make sure some other VM is the grantor
    Host.getHost(0).getVM(0).invoke("become lock grantor", () -> {
      TXLockService.createDTLS(system);
      TXLockService vm0dtls = TXLockService.getDTLS();
      DLockService vm0dlock = ((TXLockServiceImpl) vm0dtls).getInternalDistributedLockService();
      vm0dlock.becomeLockGrantor();
    });

    TXLockService.createDTLS(system);
    checkDLockRecoverGrantorMessageProcessor();

    /*
     * call TXRecoverGrantorMessageProcessor.process directly to make sure that correct behavior
     * occurs
     */

    // get txLock and hold it
    final List regionLockReqs = new ArrayList();

    Map<Object, Boolean> keymap = new HashMap<Object, Boolean>();
    keymap.put("KEY-1", true);
    keymap.put("KEY-2", true);
    keymap.put("KEY-3", true);
    keymap.put("KEY-4", true);
    regionLockReqs
        .add(new TXRegionLockRequestImpl(SEPARATOR + "testTXRecoverGrantorMessageProcessor2",
            keymap));

    TXLockService dtls = TXLockService.getDTLS();
    TXLockId txLockId = dtls.txLock(regionLockReqs, Collections.EMPTY_SET);

    final DLockService dlock = ((TXLockServiceImpl) dtls).getInternalDistributedLockService();

    // GEODE-2024: now cause grantor migration while holding the recoveryReadLock.
    // It will lock up in TXRecoverGrantorMessageProcessor until the recoveryReadLock
    // is released. Demonstrate that dtls.release() does not block forever and releases the
    // recoveryReadLock
    // allowing grantor migration to finish

    // create an observer that will block recovery messages from being processed
    MessageObserver observer = new MessageObserver();
    DistributionMessageObserver.setInstance(observer);

    try {
      System.out.println("starting thread to take over being lock grantor from vm0");

      // become the grantor - this will block waiting for a reply to the message blocked by the
      // observer
      Thread thread = new Thread(() -> {
        dlock.becomeLockGrantor();
      });
      thread.setName("TXLockServiceDUnitTest thread2");
      thread.setDaemon(true);
      thread.start();

      await("waiting for recovery to begin").until(() -> {
        return observer.isPreventingProcessing();
      });


      // spawn a thread that will unblock message processing
      // so that TXLockServiceImpl's "recovering" variable will be set
      System.out.println("starting a thread to unblock recovery in 5 seconds");
      Thread unblockThread = new Thread(() -> {
        try {
          Thread.sleep(5000);
        } catch (InterruptedException e) {
          throw new RuntimeException("sleep interrupted");
        }
        System.out.println("releasing block of recovery message processing");
        observer.releasePreventionOfProcessing();
      });
      unblockThread.setName("TXLockServiceDUnitTest unblockThread");
      unblockThread.setDaemon(true);
      unblockThread.start();

      // release txLock - this will block until unblockThread tells the observer
      // that it can process its message. Then it should release the recovery read-lock
      // allowing the grantor to finish recovery
      System.out.println("releasing transaction locks, which should block for a bit");
      dtls.release(txLockId);

      await("waiting for recovery to finish").until(() -> {
        return !((TXLockServiceImpl) dtls).isRecovering();
      });
    } finally {
      observer.releasePreventionOfProcessing();
      DistributionMessageObserver.setInstance(null);
    }
  }

  static class MessageObserver extends DistributionMessageObserver {
    final boolean[] preventingMessageProcessing = new boolean[] {false};
    final boolean[] preventMessageProcessing = new boolean[] {true};


    public boolean isPreventingProcessing() {
      synchronized (preventingMessageProcessing) {
        return preventingMessageProcessing[0];
      }
    }

    public void releasePreventionOfProcessing() {
      synchronized (preventMessageProcessing) {
        preventMessageProcessing[0] = false;
      }
    }

    @Override
    public void beforeProcessMessage(ClusterDistributionManager dm, DistributionMessage message) {
      if (message instanceof DLockRecoverGrantorMessage) {
        synchronized (preventingMessageProcessing) {
          preventingMessageProcessing[0] = true;
        }
        synchronized (preventMessageProcessing) {
          while (preventMessageProcessing[0]) {
            try {
              preventMessageProcessing.wait(50);
            } catch (InterruptedException e) {
              throw new RuntimeException("sleep interrupted");
            }
          }
        }
      }
    }

  }


  protected static volatile TXLockId testTXLock_TXLockId;

  @Test
  public void testTXLock() {
    LogWriterUtils.getLogWriter().info("[testTXLock]");
    final int grantorVM = 0;
    final int clientA = 1;
    final int clientB = 2;
    final Set participants = Collections.EMPTY_SET;
    final List regionLockReqs = new ArrayList();

    Map<Object, Boolean> keymap1 = new HashMap<Object, Boolean>();
    keymap1.put("KEY-1", true);
    keymap1.put("KEY-2", true);
    keymap1.put("KEY-3", true);
    keymap1.put("KEY-4", true);
    Map<Object, Boolean> keymap2 = new HashMap<Object, Boolean>();
    keymap2.put("KEY-A", true);
    keymap2.put("KEY-B", true);
    keymap2.put("KEY-C", true);
    keymap2.put("KEY-D", true);

    regionLockReqs.add(new TXRegionLockRequestImpl(SEPARATOR + "testTXLock1", keymap1));
    regionLockReqs.add(new TXRegionLockRequestImpl(SEPARATOR + "testTXLock2", keymap2));

    // create grantor
    LogWriterUtils.getLogWriter().info("[testTXLock] create grantor");

    Host.getHost(0).getVM(grantorVM).invoke(new SerializableRunnable() {
      @Override
      public void run() {
        TXLockService.createDTLS(system);
      }
    });
    sleep(20);

    // create client and request txLock
    LogWriterUtils.getLogWriter().info("[testTXLock] create clientA and request txLock");

    Host.getHost(0).getVM(clientA).invoke(new SerializableRunnable() {
      @Override
      public void run() {
        TXLockService.createDTLS(system);
      }
    });

    Host.getHost(0).getVM(clientA)
        .invoke(new SerializableRunnable("[testTXLock] create clientA and request txLock") {
          @Override
          public void run() {
            TXLockService dtls = TXLockService.getDTLS();
            testTXLock_TXLockId = dtls.txLock(regionLockReqs, participants);
            assertNotNull("testTXLock_TXLockId is null", testTXLock_TXLockId);
          }
        });

    // create nuther client and request overlapping txLock... verify fails
    LogWriterUtils.getLogWriter().info("[testTXLock] create clientB and fail txLock");

    Host.getHost(0).getVM(clientB).invoke(new SerializableRunnable() {
      @Override
      public void run() {
        TXLockService.createDTLS(system);
      }
    });

    Host.getHost(0).getVM(clientB).invoke(new SerializableRunnable() {
      @Override
      public void run() {
        try {
          TXLockService dtls = TXLockService.getDTLS();
          dtls.txLock(regionLockReqs, participants);
          fail("expected CommitConflictException");
        } catch (CommitConflictException expected) {
        }
      }
    });

    /*
     * try { Host.getHost(0).getVM(clientB).invoke(() -> TXLockServiceDUnitTest.txLock_DTLS(
     * regionLockReqs, participants )); fail("expected CommitConflictException"); } catch
     * (RMIException expected) { assertTrue(expected.getCause() instanceof CommitConflictException);
     * }
     */

    // release txLock
    LogWriterUtils.getLogWriter().info("[testTXLock] clientA releases txLock");

    Host.getHost(0).getVM(clientA)
        .invoke(new SerializableRunnable("[testTXLock] clientA releases txLock") {
          @Override
          public void run() {
            TXLockService dtls = TXLockService.getDTLS();
            dtls.release(testTXLock_TXLockId);
          }
        });
    sleep(20);

    // try nuther client again and verify success
    LogWriterUtils.getLogWriter().info("[testTXLock] clientB requests txLock");

    Host.getHost(0).getVM(clientB)
        .invoke(new SerializableRunnable("[testTXLock] clientB requests txLock") {
          @Override
          public void run() {
            TXLockService dtls = TXLockService.getDTLS();
            testTXLock_TXLockId = dtls.txLock(regionLockReqs, participants);
            assertNotNull("testTXLock_TXLockId is null", testTXLock_TXLockId);
          }
        });

    // release txLock
    LogWriterUtils.getLogWriter().info("[testTXLock] clientB releases txLock");

    Host.getHost(0).getVM(clientB)
        .invoke(new SerializableRunnable("[testTXLock] clientB releases txLock") {
          @Override
          public void run() {
            TXLockService dtls = TXLockService.getDTLS();
            dtls.release(testTXLock_TXLockId);
          }
        });
  }

  protected static volatile TXLockId testTXOriginatorRecoveryProcessor_TXLockId;

  @Test
  public void testTXOriginatorRecoveryProcessor() {
    LogWriterUtils.getLogWriter().info("[testTXOriginatorRecoveryProcessor]");
    final int originatorVM = 0;
    final int grantorVM = 1;
    final int particpantA = 2;
    final int particpantB = 3;

    final List regionLockReqs = new ArrayList();

    Map<Object, Boolean> keymap1 = new HashMap<Object, Boolean>();
    keymap1.put("KEY-1", true);
    keymap1.put("KEY-2", true);
    keymap1.put("KEY-3", true);
    keymap1.put("KEY-4", true);

    regionLockReqs
        .add(new TXRegionLockRequestImpl(SEPARATOR + "testTXOriginatorRecoveryProcessor", keymap1));

    // build participants set...
    InternalDistributedMember dmId = null;
    final Set participants = new HashSet();
    for (int i = 1; i <= particpantB; i++) {
      final int finalvm = i;
      dmId = (InternalDistributedMember) Host.getHost(0).getVM(finalvm)
          .invoke(() -> TXLockServiceDUnitTest.fetchDistributionManagerId());
      assertEquals("dmId should not be null for vm " + finalvm, false, dmId == null);
      participants.add(dmId);
    }

    // create grantor
    LogWriterUtils.getLogWriter()
        .info("[testTXOriginatorRecoveryProcessor] grantorVM becomes grantor");

    Host.getHost(0).getVM(grantorVM).invoke(new SerializableRunnable() {
      @Override
      public void run() {
        TXLockService.createDTLS(system);
      }
    });

    Host.getHost(0).getVM(grantorVM)
        .invoke(() -> TXLockServiceDUnitTest.identifyLockGrantor_DTLS());

    Boolean isGrantor = (Boolean) Host.getHost(0).getVM(grantorVM)
        .invoke(() -> TXLockServiceDUnitTest.isLockGrantor_DTLS());
    assertEquals("isLockGrantor should not be false for DTLS", Boolean.TRUE, isGrantor);

    // have a originatorVM get a txLock with three participants including grantor
    LogWriterUtils.getLogWriter()
        .info("[testTXOriginatorRecoveryProcessor] originatorVM requests txLock");

    Host.getHost(0).getVM(originatorVM).invoke(new SerializableRunnable() {
      @Override
      public void run() {
        TXLockService.createDTLS(system);
      }
    });
    Host.getHost(0).getVM(originatorVM).invoke(new SerializableRunnable(
        "[testTXOriginatorRecoveryProcessor] originatorVM requests txLock") {
      @Override
      public void run() {
        TXLockService dtls = TXLockService.getDTLS();
        testTXOriginatorRecoveryProcessor_TXLockId = dtls.txLock(regionLockReqs, participants);
        assertNotNull("testTXOriginatorRecoveryProcessor_TXLockId is null",
            testTXOriginatorRecoveryProcessor_TXLockId);
      }
    });

    // create dtls in each participant
    Host.getHost(0).getVM(particpantA).invoke(new SerializableRunnable() {
      @Override
      public void run() {
        TXLockService.createDTLS(system);
      }
    });
    Host.getHost(0).getVM(particpantB).invoke(new SerializableRunnable() {
      @Override
      public void run() {
        TXLockService.createDTLS(system);
      }
    });

    // disconnect originatorVM without releasing txLock
    /*
     * doesn't currently trigger the DLockLessorDepatureHandler... TODO
     * Host.getHost(0).getVM(originatorVM).invoke(new SerializableRunnable() { public void run() {
     * TXLockService.destroyServices(); } });
     */

    /*
     * Host.getHost(0).getVM(originatorVM).invoke(new SerializableRunnable() { public void run() {
     * InternalDistributedSystem sys = (InternalDistributedSystem)
     * InternalDistributedSystem.getAnyInstance(); if (sys != null) { sys.disconnect(); } } });
     */

    Host.getHost(0).getVM(originatorVM).invoke(new SerializableRunnable() {
      @Override
      public void run() {
        TXLockService.destroyServices();
      }
    });
    Host.getHost(0).getVM(originatorVM).invoke(() -> disconnectFromDS());

    // grantor sends TXOriginatorRecoveryMessage...
    // TODO: verify processing of message? and have test sleep until finished
    sleep(200);

    // verify txLock is released...
    Host.getHost(0).getVM(particpantA).invoke(
        new SerializableRunnable("[testTXOriginatorRecoveryProcessor] verify txLock is released") {
          @Override
          public void run() {
            TXLockService dtls = TXLockService.getDTLS();
            testTXOriginatorRecoveryProcessor_TXLockId = dtls.txLock(regionLockReqs, participants);
            assertNotNull("testTXOriginatorRecoveryProcessor_TXLockId is null",
                testTXOriginatorRecoveryProcessor_TXLockId);
          }
        });

    Host.getHost(0).getVM(particpantA).invoke(new SerializableRunnable(
        "[testTXOriginatorRecoveryProcessor] particpantA releases txLock") {
      @Override
      public void run() {
        TXLockService dtls = TXLockService.getDTLS();
        dtls.release(testTXOriginatorRecoveryProcessor_TXLockId);
      }
    });
  }

  @Test
  public void testDTLSIsDistributed() {
    LogWriterUtils.getLogWriter().info("[testDTLSIsDistributed]");

    // have all vms lock and hold the same LTLS lock simultaneously
    final Host host = Host.getHost(0);
    int vmCount = host.getVMCount();
    for (int vm = 0; vm < vmCount; vm++) {
      final int finalvm = vm;
      LogWriterUtils.getLogWriter().info("[testDTLSIsDistributed] testing vm " + finalvm);

      Host.getHost(0).getVM(finalvm).invoke(new SerializableRunnable() {
        @Override
        public void run() {
          TXLockService.createDTLS(system);
        }
      });

      // assert that isDistributed returns false
      Boolean isDistributed =
          (Boolean) host.getVM(finalvm).invoke(() -> TXLockServiceDUnitTest.isDistributed_DTLS());
      assertEquals("isDistributed should be true for DTLS", Boolean.TRUE, isDistributed);
      LogWriterUtils.getLogWriter().info("[testDTLSIsDistributed] isDistributed=" + isDistributed);

      // lock a key...
      Boolean gotLock =
          (Boolean) host.getVM(finalvm).invoke(() -> TXLockServiceDUnitTest.lock_DTLS("KEY"));
      assertEquals("gotLock is false after calling lock_DTLS", Boolean.TRUE, gotLock);
      LogWriterUtils.getLogWriter().info("[testDTLSIsDistributed] gotLock=" + gotLock);

      // unlock it...
      Boolean unlock =
          (Boolean) host.getVM(finalvm).invoke(() -> TXLockServiceDUnitTest.unlock_DTLS("KEY"));
      assertEquals("unlock is false after calling unlock_DTLS", Boolean.TRUE, unlock);
      LogWriterUtils.getLogWriter().info("[testDTLSIsDistributed] unlock=" + unlock);
    }
  }

  // -------------------------------------------------------------------------
  // Static support methods
  // -------------------------------------------------------------------------

  public static void remoteDumpAllDLockServices() {
    DLockService.dumpAllServices();
  }

  /**
   * Creates a new DistributedLockService in a remote VM.
   *
   * @param name The name of the newly-created DistributedLockService. It is recommended that the
   *        name of the Region be the {@link #getUniqueName()} of the test, or at least derive from
   *        it.
   */
  protected static void remoteCreateService(String name) {
    DistributedLockService newService = DistributedLockService.create(name, system);
    logInfo("Created " + newService);
  }

  private static void logInfo(String msg) {
    system.getLogWriter().info(msg);
  }

  private static void sleep(long millis) {
    try {
      Thread.sleep(millis);
    } catch (InterruptedException ex) {
      fail("interrupted");
    }
  }

  /**
   * Connects a DistributedSystem, saves it in static variable "system"
   */
  private static void connectDistributedSystem() {
    system = (new TXLockServiceDUnitTest()).getSystem();
  }

  private static InternalDistributedMember identifyLockGrantor(String serviceName) {
    DLockService service = (DLockService) DistributedLockService.getServiceNamed(serviceName);
    assertNotNull(service);

    InternalDistributedMember grantor = service.getLockGrantorId().getLockGrantorMember();
    assertNotNull(grantor);
    logInfo("In identifyLockGrantor - grantor is " + grantor);
    return grantor;
  }

  /**
   * Accessed via reflection. DO NOT REMOVE
   */
  protected static InternalDistributedMember identifyLockGrantor_DTLS() {
    TXLockService dtls = TXLockService.getDTLS();
    String serviceName = ((TXLockServiceImpl) dtls).getInternalDistributedLockService().getName();
    return identifyLockGrantor(serviceName);
  }

  /**
   * Accessed via reflection. DO NOT REMOVE
   */
  protected static Boolean lock_DTLS(Object key) {
    TXLockService dtls = TXLockService.getDTLS();
    boolean gotLock =
        ((TXLockServiceImpl) dtls).getInternalDistributedLockService().lock(key, -1, -1);
    logInfo("lock_LTLS gotLock (hopefully true): " + gotLock);
    return Boolean.valueOf(gotLock);
  }

  /**
   * Accessed via reflection. DO NOT REMOVE.
   */
  protected static Boolean isLockGrantor_DTLS() {
    TXLockService dtls = TXLockService.getDTLS();

    if (true) {
      DLockService service = DLockService.getInternalServiceNamed(
          ((TXLockServiceImpl) dtls).getInternalDistributedLockService().getName());
      assertNotNull(service);

      assertEquals("DTLS and DLock should both report same isLockGrantor result", true,
          dtls.isLockGrantor() == service.isLockGrantor());
    }

    Boolean result = Boolean.valueOf(dtls.isLockGrantor());
    logInfo("isLockGrantor_DTLS: " + result);
    return result;
  }

  /**
   * Accessed via reflection. DO NOT REMOVE
   */
  protected static Boolean isDistributed_DTLS() {
    TXLockService dtls = TXLockService.getDTLS();
    boolean isDistributed =
        ((TXLockServiceImpl) dtls).getInternalDistributedLockService().isDistributed();

    DLockService svc = ((TXLockServiceImpl) dtls).getInternalDistributedLockService();
    assertNotNull(svc);
    assertEquals("DTLS InternalDistributedLockService should not be destroyed", false,
        svc.isDestroyed());

    // sleep(50);

    if (true) {
      DLockService service = DLockService.getInternalServiceNamed(svc.getName());
      assertNotNull(service);

      assertEquals("DTLS and DLock should both report same isDistributed result", true,
          isDistributed == service.isDistributed());
    }

    Boolean result = Boolean.valueOf(isDistributed);
    logInfo("isDistributed_DTLS (hopefully true): " + result);
    return result;
  }

  // private static void becomeLockGrantor(String serviceName) {
  // InternalDistributedLockService service = (InternalDistributedLockService)
  // DistributedLockService.getServiceNamed(serviceName);
  // assertNotNull(service);
  // logInfo("About to call becomeLockGrantor...");
  // service.becomeLockGrantor();
  // }

  /**
   * Accessed via reflection. DO NOT REMOVE
   */
  protected static void checkGetAndDestroy() {
    assertNull(TXLockService.getDTLS());

    TXLockService dtls = TXLockService.createDTLS(system);
    assertNotNull(dtls);
    assertEquals(true, dtls == TXLockService.getDTLS());
    assertEquals(false, dtls.isDestroyed());

    TXLockService.destroyServices();
    assertEquals(true, dtls.isDestroyed());
    assertNull(TXLockService.getDTLS());

    dtls = TXLockService.createDTLS(system);
    assertNotNull(dtls);
    assertEquals(true, dtls == TXLockService.getDTLS());
    assertEquals(false, dtls.isDestroyed());
  }

  /**
   * Accessed via reflection. DO NOT REMOVE
   */
  protected static Boolean unlock_DTLS(Object key) {
    TXLockService dtls = TXLockService.getDTLS();
    try {
      ((TXLockServiceImpl) dtls).getInternalDistributedLockService().unlock(key);
      return Boolean.TRUE;
    } catch (Exception e) {
      return Boolean.FALSE;
    }
  }


  /**
   * Accessed via reflection. DO NOT REMOVE.
   */
  protected static InternalDistributedMember fetchDistributionManagerId() {
    InternalDistributedSystem sys = InternalDistributedSystem.getAnyInstance();
    if (sys != null) {
      return sys.getDistributionManager().getId();
    } else {
      return null;
    }
  }

  private static void destroyServices() {
    TXLockService.destroyServices();
  }

  // -------------------------------------------------------------------------
  // Non-static support methods
  // -------------------------------------------------------------------------

  /**
   * Assumes there is only one host, and invokes the given method in the first numVMs VMs that host
   * knows about.
   */
  public void forNumVMsInvoke(int numVMs, String methodName, Object[] args) {
    Host host = Host.getHost(0);
    for (int i = 0; i < numVMs; i++) {
      logInfo("Invoking " + methodName + "on VM#" + i);
      host.getVM(i).invoke(this.getClass(), methodName, args);
    }
  }

  public void forEachVMInvoke(String methodName, Object[] args) {
    Host host = Host.getHost(0);
    int vmCount = host.getVMCount();
    for (int i = 0; i < vmCount; i++) {
      LogWriterUtils.getLogWriter().info("Invoking " + methodName + "on VM#" + i);
      host.getVM(i).invoke(this.getClass(), methodName, args);
    }
  }

  @Override
  public Properties getDistributedSystemProperties() {
    Properties props = super.getDistributedSystemProperties();
    props.setProperty(LOG_LEVEL, LogWriterUtils.getDUnitLogLevel());
    return props;
  }

  // private synchronized void assertGrantorIsConsistent(Serializable id) {
  // if (this.lockGrantor == null) {
  // this.lockGrantor = id;
  // } else {
  // assertIndexDetailsEquals("assertGrantorIsConsistent failed", lockGrantor, id);
  // }
  // }

  // private void distributedCreateService(int numVMs, String serviceName) {
  // forEachVMInvoke(
  // "remoteCreateService",
  // new Object[] { serviceName });
  //
  // remoteCreateService(serviceName);
  // }

  private void checkDLockRecoverGrantorMessageProcessor() {
    /*
     * simple test to make sure getDLockRecoverGrantorMessageProcessor returns instance of
     * TXRecoverGrantorMessageProcessor
     */
    DLockService dlock = null;

    TXLockServiceImpl dtls = (TXLockServiceImpl) TXLockService.getDTLS();
    assertNotNull("DTLS should not be null", dtls);
    dlock = dtls.getInternalDistributedLockService();
    assertEquals("DTLS should use TXRecoverGrantorMessageProcessor", true,
        dlock.getDLockRecoverGrantorMessageProcessor() instanceof TXRecoverGrantorMessageProcessor);
  }

  // -------------------------------------------------------------------------
  // Inner class
  // -------------------------------------------------------------------------

  private static class TestDLockRecoverGrantorProcessor extends ReplyProcessor21 {
    public TestDLockRecoverGrantorProcessor(DistributionManager dm, Set members) {
      super(dm.getSystem(), members);
    }

    @Override
    protected boolean allowReplyFromSender() {
      return true;
    }

    @Override
    public void process(DistributionMessage msg) {
      DLockRecoverGrantorProcessor.DLockRecoverGrantorReplyMessage reply =
          (DLockRecoverGrantorProcessor.DLockRecoverGrantorReplyMessage) msg;
      testTXRecoverGrantor_replyCode_PASS =
          (reply.getReplyCode() == DLockRecoverGrantorProcessor.DLockRecoverGrantorReplyMessage.OK);
      testTXRecoverGrantor_heldLocks_PASS = (reply.getHeldLocks().length == 0);
    }
  }

}