GEODE-8492: fix redis 'clients' statistic (apache#5510)

The redis "clients" statistics went negative because it was being decremented multiple
times when a connection disconnected. It is now only decremented once.

geode-redis/src/integrationTest/java/org/apache/geode/redis/internal/RedisStatsIntegrationTest.java

@@ -0,0 +1,43 @@
+/*
+ * 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.redis.internal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.ClassRule;
+import org.junit.Test;
+import redis.clients.jedis.Jedis;
+
+import org.apache.geode.redis.GeodeRedisServerRule;
+import org.apache.geode.test.awaitility.GeodeAwaitility;
+
+public class RedisStatsIntegrationTest {
+
+  @ClassRule
+  public static GeodeRedisServerRule server = new GeodeRedisServerRule();
+
+  @Test
+  public void clientsStat_withConnectAndClose_isCorrect() throws InterruptedException {
+    long initialClients = server.getServer().getStats().getClients();
+    Jedis jedis = new Jedis("localhost", server.getPort(), 10000000);
+
+    jedis.ping();
+    assertThat(server.getServer().getStats().getClients()).isEqualTo(initialClients + 1);
+
+    jedis.close();
+    GeodeAwaitility.await().untilAsserted(
+        () -> assertThat(server.getServer().getStats().getClients()).isEqualTo(initialClients));
+  }
+}

geode-redis/src/main/java/org/apache/geode/redis/internal/GeodeRedisServer.java

@@ -16,6 +16,7 @@
 
 import org.apache.logging.log4j.Logger;
 
+import org.apache.geode.annotations.VisibleForTesting;
 import org.apache.geode.cache.Cache;
 import org.apache.geode.cache.Region;
 import org.apache.geode.distributed.internal.InternalDistributedSystem;
@@ -114,6 +115,12 @@ private static RedisStats createStats(InternalCache cache) {
     return new RedisStats(system.getStatisticsManager(), statisticsClock);
   }
 
+
+  @VisibleForTesting
+  RedisStats getStats() {
+    return redisStats;
+  }
+
   /**
    * Precedence of the internal property overrides the global system property.
    */

geode-redis/src/main/java/org/apache/geode/redis/internal/RedisStats.java

@@ -25,6 +25,7 @@
 import org.apache.geode.StatisticsType;
 import org.apache.geode.StatisticsTypeFactory;
 import org.apache.geode.annotations.Immutable;
+import org.apache.geode.annotations.VisibleForTesting;
 import org.apache.geode.internal.statistics.StatisticsClock;
 import org.apache.geode.internal.statistics.StatisticsTypeFactoryImpl;
 
@@ -177,6 +178,11 @@ public void removeClient() {
     stats.incLong(clientId, -1);
   }
 
+  @VisibleForTesting
+  long getClients() {
+    return stats.getLong(clientId);
+  }
+
   public long startPassiveExpirationCheck() {
     return getTime();
   }

geode-redis/src/main/java/org/apache/geode/redis/internal/netty/ExecutionHandlerContext.java

@@ -19,6 +19,7 @@
 import java.io.IOException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 
@@ -76,6 +77,7 @@ public class ExecutionHandlerContext extends ChannelInboundHandlerAdapter {
   private final Runnable shutdownInvoker;
   private final RedisStats redisStats;
   private final EventLoopGroup subscriberGroup;
+  private final AtomicBoolean channelInactive = new AtomicBoolean();
   private final int MAX_QUEUED_COMMANDS =
       Integer.getInteger("geode.redis.commandQueueSize", 1000);
   private final LinkedBlockingQueue<Command> commandQueue =
@@ -241,13 +243,16 @@ private RedisResponse getExceptionResponse(ChannelHandlerContext ctx, Throwable
 
   @Override
   public void channelInactive(ChannelHandlerContext ctx) {
-    if (logger.isDebugEnabled()) {
-      logger.debug("GeodeRedisServer-Connection closing with " + ctx.channel().remoteAddress());
+    if (channelInactive.compareAndSet(false, true)) {
+      if (logger.isDebugEnabled()) {
+        logger.debug("GeodeRedisServer-Connection closing with " + ctx.channel().remoteAddress());
+      }
+
+      commandQueue.offer(TERMINATE_COMMAND);
+      redisStats.removeClient();
+      ctx.channel().close();
+      ctx.close();
     }
-    commandQueue.offer(TERMINATE_COMMAND);
-    redisStats.removeClient();
-    ctx.channel().close();
-    ctx.close();
   }
 
   private void executeCommand(Command command) {