@Test
  public void dataWriterCannotRegisterInterestAcrossFailover() throws Exception {
    Properties props = getVMPropertiesWithPermission("dataWrite");
    if (shouldSetObjectFilterProperty(clientVmConfiguration)) {
      props.setProperty(ConfigurationProperties.SERIALIZABLE_OBJECT_FILTER,
          "org.apache.geode.security.templates.UsernamePrincipal");
    }

    int server1Port = server1.getPort();
    int server2Port = server2.getPort();

    ClientVM client1 = csRule.startClientVM(3, clientVmConfiguration, props,
        cf -> cf.addPoolServer("localhost", server1Port)
            .addPoolServer("localhost", server2Port)
            .setPoolSubscriptionEnabled(true)
            .setPoolSubscriptionRedundancy(2));

    // Initialize cache
    client1.invoke(() -> {
      ClientCache cache = ClusterStartupRule.getClientCache();
      ClientRegionFactory<String, String> rf =
          cache.createClientRegionFactory(ClientRegionShortcut.PROXY);
      Region<String, String> region1 = rf.create(regionName);
    });

    ClientVM client = client1;

    // Client should be able to register and unregister interests.
    client.invoke(() -> {
      Region<String, String> region = ClusterStartupRule.getClientCache().getRegion(regionName);
      assertThatThrownBy(() -> region.registerInterest(client_k1))
          .hasCauseInstanceOf(NotAuthorizedException.class);
      assertThatThrownBy(() -> region.registerInterestRegex("client-.*"))
          .hasCauseInstanceOf(NotAuthorizedException.class);
      // Attempts to unregister will fail client-side. The client maintains its own lists of
      // interests and, since the above failed, any unregistering of interest will prematurely
      // terminate before contacting any server. No authorization is attempted.
    });

    // Initialize client cache and region. Get the port of the primary connected server.
    MemberVM server_to_fail = determinePrimaryServer(client);

    // Bring down primary server
    server_to_fail.stop(true);

    // Confirm failover
    MemberVM secondaryServer = (server1.getPort() == server_to_fail.getPort()) ? server2 : server1;
    await().until(() -> getPrimaryServerPort(client) == secondaryServer.getPort());

    // Confirm permissions.
    client.invoke(() -> {
      Region<String, String> region = ClusterStartupRule.getClientCache().getRegion(regionName);
      assertThatThrownBy(() -> region.registerInterest(client_k1))
          .hasCauseInstanceOf(NotAuthorizedException.class);
      assertThatThrownBy(() -> region.registerInterestRegex("client-.*"))
          .hasCauseInstanceOf(NotAuthorizedException.class);
    });
  }
This test class reproduces the tests present in
{@link ClientDataAuthorizationUsingLegacySecurityDUnitTest} and confirms that permissions are
maintained over failover.
geode-core/src/upgradeTest/java/org/apache/geode/security/ClientDataAuthorizationUsingLegacySecurityWithFailoverDUnitTest.java