[Bug] ZooKeeper user entry not deleted when KafkaUser is deleted
When a new `KafkaUser` resource is created with quotas, a new user entry is created in ZooKeeper in the `/config/users` znode (i.e. `/config/users/my-user`). It contains a JSON with `version` and `config` fields.
When the `KafkaUser` resource is deleted, the above entry isn't deleted but the `config` field is emptied.
Because the `KafkaUser` isn't there anymore, the user entry should be completely deleted.
diff --git a/systemtest/src/test/java/io/strimzi/systemtest/UserST.java b/systemtest/src/test/java/io/strimzi/systemtest/UserST.java
index 5c2d2f93b..c5bde6605 100644
--- a/systemtest/src/test/java/io/strimzi/systemtest/UserST.java
+++ b/systemtest/src/test/java/io/strimzi/systemtest/UserST.java
@@ -171,12 +171,19 @@ class UserST extends BaseST {
         assertThat(result.out().contains("producer_byte_rate=" + prodRate), is(true));
         assertThat(result.out().contains("consumer_byte_rate=" + consRate), is(true));
 
+        String zkListCommand = "sh /opt/kafka/bin/zookeeper-shell.sh localhost:21810 <<< 'ls /config/users'";
+        ExecResult zkResult = cmdKubeClient().execInPod(KafkaResources.zookeeperPodName(CLUSTER_NAME, 0), "/bin/bash", "-c", zkListCommand);
+        assertThat(zkResult.out().contains(userName), is(true));
+
+        // delete user
         KafkaUserResource.kafkaUserClient().inNamespace(NAMESPACE).withName(userName).delete();
         KafkaUserUtils.waitForKafkaUserDeletion(userName);
 
         ExecResult resultAfterDelete = cmdKubeClient().execInPod(KafkaResources.kafkaPodName(CLUSTER_NAME, 0), "/bin/bash", "-c", command);
         assertThat(resultAfterDelete.out(), emptyString());
 
+        ExecResult zkDeleteResult = cmdKubeClient().execInPod(KafkaResources.zookeeperPodName(CLUSTER_NAME, 0), "/bin/bash", "-c", zkListCommand);
+        assertThat(zkDeleteResult.out().contains(userName), is(false));
     }
 
     void createBigAmountOfUsers(String typeOfUser) {
diff --git a/user-operator/src/main/java/io/strimzi/operator/user/operator/KafkaUserQuotasOperator.java b/user-operator/src/main/java/io/strimzi/operator/user/operator/KafkaUserQuotasOperator.java
index 48aaf1028..5d4126c2f 100644
--- a/user-operator/src/main/java/io/strimzi/operator/user/operator/KafkaUserQuotasOperator.java
+++ b/user-operator/src/main/java/io/strimzi/operator/user/operator/KafkaUserQuotasOperator.java
@@ -195,6 +195,12 @@ public class KafkaUserQuotasOperator {
         return false;
     }
 
+    private boolean configJsonIsEmpty(JsonObject json) {
+        validateJsonVersion(json);
+        JsonObject config = json.getJsonObject("config");
+        return config.isEmpty();
+    }
+
     /**
      * Delete the quotas for the given user.
      * It is not an error if the user doesn't exist, or doesn't currently have any quotas.
@@ -206,7 +212,13 @@ public class KafkaUserQuotasOperator {
 
         if (data != null)   {
             log.debug("Deleting quotas for user {}", username);
-            zkClient.writeData("/config/users/" + username, removeQuotasFromJsonUser(data));
+            JsonObject deleteJson = removeQuotasFromJsonUser(data);
+            if (configJsonIsEmpty(deleteJson)) {
+                zkClient.deleteRecursive("/config/users/" + username);
+                log.debug("User {} deleted from ZK store", username);
+            } else {
+                zkClient.writeData("/config/users/" + username, deleteJson.toBuffer().getBytes());
+            }
             notifyChanges(username);
         } else {
             log.warn("Quotas for user {} already don't exist", username);
@@ -218,9 +230,9 @@ public class KafkaUserQuotasOperator {
      *
      * @param user JSON string with existing user configuration as byte[]
      *
-     * @return  Returns the updated JSON without the quotas as byte array
+     * @return  Returns the updated JSON without the quotas
      */
-    protected byte[] removeQuotasFromJsonUser(byte[] user)   {
+    protected JsonObject removeQuotasFromJsonUser(byte[] user)   {
         JsonObject json = new JsonObject(new String(user, StandardCharsets.UTF_8));
 
         validateJsonVersion(json);
@@ -238,8 +250,7 @@ public class KafkaUserQuotasOperator {
                 config.remove("request_percentage");
             }
         }
-
-        return json.encode().getBytes(StandardCharsets.UTF_8);
+        return json;
     }
 
     protected void validateJsonVersion(JsonObject json) {
diff --git a/user-operator/src/main/java/io/strimzi/operator/user/operator/ScramShaCredentials.java b/user-operator/src/main/java/io/strimzi/operator/user/operator/ScramShaCredentials.java
index 151a59567..051b098c6 100644
--- a/user-operator/src/main/java/io/strimzi/operator/user/operator/ScramShaCredentials.java
+++ b/user-operator/src/main/java/io/strimzi/operator/user/operator/ScramShaCredentials.java
@@ -56,6 +56,12 @@ public class ScramShaCredentials {
         notifyChanges(username);
     }
 
+    private boolean configJsonIsEmpty(JsonObject json) {
+        validateJsonVersion(json);
+        JsonObject config = json.getJsonObject("config");
+        return config.isEmpty();
+    }
+
     /**
      * Delete the SCRAM-SHA credentials for the given user.
      * It is not an error if the user doesn't exist, or doesn't currently have any SCRAM-SHA credentials.
@@ -67,7 +73,12 @@ public class ScramShaCredentials {
 
         if (data != null)   {
             log.debug("Deleting {} credentials for user {}", mechanism.mechanismName(), username);
-            zkClient.writeData("/config/users/" + username, deleteUserJson(data));
+            JsonObject deletedJson = deleteUserJson(data);
+            if (configJsonIsEmpty(deletedJson)) {
+                zkClient.deleteRecursive("/config/users/" + username);
+            } else {
+                zkClient.writeData("/config/users/" + username, deletedJson.toBuffer().getBytes());
+            }
             notifyChanges(username);
         } else {
             log.warn("Credentials for user {} already don't exist", username);
@@ -154,6 +165,11 @@ public class ScramShaCredentials {
         }
     }
 
+    /* test */
+    boolean isPathExist(String path)    {
+        return zkClient.exists(path);
+    }
+
     /**
      * Generates the JSON with the credentials
      *
@@ -209,9 +225,9 @@ public class ScramShaCredentials {
      *
      * @param user JSON string with existing user configuration as byte[]
      *
-     * @return  Returns the updated JSON without the SCRAM credentials as byte array
+     * @return  Returns the updated JSON without the SCRAM credentials
      */
-    protected byte[] deleteUserJson(byte[] user)   {
+    protected JsonObject deleteUserJson(byte[] user)   {
         JsonObject json = new JsonObject(new String(user, Charset.defaultCharset()));
 
         validateJsonVersion(json);
@@ -224,7 +240,7 @@ public class ScramShaCredentials {
             json.getJsonObject("config").remove(mechanism.mechanismName());
         }
 
-        return json.encode().getBytes(Charset.defaultCharset());
+        return json;
     }
 
     protected void validateJsonVersion(JsonObject json)   {
diff --git a/user-operator/src/test/java/io/strimzi/operator/user/operator/KafkaUserQuotasIT.java b/user-operator/src/test/java/io/strimzi/operator/user/operator/KafkaUserQuotasIT.java
index 61adfafd8..63ceb649b 100644
--- a/user-operator/src/test/java/io/strimzi/operator/user/operator/KafkaUserQuotasIT.java
+++ b/user-operator/src/test/java/io/strimzi/operator/user/operator/KafkaUserQuotasIT.java
@@ -14,7 +14,7 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
-import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
 
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.CoreMatchers.notNullValue;
@@ -84,6 +84,7 @@ public class KafkaUserQuotasIT {
         assertThat(kuq.exists("normalDelete"), is(true));
         kuq.delete("normalDelete");
         assertThat(kuq.exists("normalDelete"), is(false));
+        assertThat(kuq.isPathExist("/config/users/normalDelete"), is(false));
     }
 
     @Test
@@ -94,6 +95,7 @@ public class KafkaUserQuotasIT {
         kuq.delete("doubleDelete");
         kuq.delete("doubleDelete");
         assertThat(kuq.exists("doubleDelete"), is(false));
+        assertThat(kuq.isPathExist("/config/users/doubleDelete"), is(false));
     }
 
     @Test
@@ -141,23 +143,26 @@ public class KafkaUserQuotasIT {
     @Test
     public void testDeletion()  {
         JsonObject original = new JsonObject().put("version", 1).put("config", kuq.quotasToJson(defaultQuotas));
-        JsonObject updated = new JsonObject(new String(kuq.removeQuotasFromJsonUser(original.encode().getBytes(Charset.defaultCharset())), Charset.defaultCharset()));
+        original.getJsonObject("config").put("persist", 42);
+        JsonObject updated = kuq.removeQuotasFromJsonUser(original.encode().getBytes(StandardCharsets.UTF_8));
         assertThat(updated.getJsonObject("config").getString("consumer_byte_rate"), is(nullValue()));
+        assertThat(updated.getJsonObject("config").getInteger("persist"), is(42));
 
         original = new JsonObject().put("version", 1).put("config", new JsonObject().put("producer_byte_rate", "1000").put("consumer_byte_rate", "2000"));
-        updated = new JsonObject(new String(kuq.removeQuotasFromJsonUser(original.encode().getBytes(Charset.defaultCharset())), Charset.defaultCharset()));
+        updated = kuq.removeQuotasFromJsonUser(original.encode().getBytes(StandardCharsets.UTF_8));
         assertThat(updated.getJsonObject("config").getString("producer_byte_rate"), is(nullValue()));
         assertThat(updated.getJsonObject("config").getString("consumer_byte_rate"), is(nullValue()));
 
         original = new JsonObject().put("version", 1).put("config", new JsonObject());
-        updated = new JsonObject(new String(kuq.removeQuotasFromJsonUser(original.encode().getBytes(Charset.defaultCharset())), Charset.defaultCharset()));
+        updated = kuq.removeQuotasFromJsonUser(original.encode().getBytes(StandardCharsets.UTF_8));
         assertThat(updated.getJsonObject("config").getString("consumer_byte_rate"), is(nullValue()));
         assertThat(updated.getJsonObject("config").getString("producer_byte_rate"), is(nullValue()));
 
         original = new JsonObject().put("version", 1).put("config", new JsonObject().put("consumer_byte_rate", "1000"));
-        updated = new JsonObject(new String(kuq.removeQuotasFromJsonUser(original.encode().getBytes(Charset.defaultCharset())), Charset.defaultCharset()));
+        updated = kuq.removeQuotasFromJsonUser(original.encode().getBytes(StandardCharsets.UTF_8));
         assertThat(updated.getJsonObject("config").getString("producer_byte_rate"), is(nullValue()));
         assertThat(updated.getJsonObject("config").getString("consumer_byte_rate"), is(nullValue()));
+
     }
 
     @Test
@@ -165,14 +170,14 @@ public class KafkaUserQuotasIT {
         JsonObject original = new JsonObject().put("version", 1).put("config", new JsonObject().put("consumer_byte_rate", "1000"));
         KafkaUserQuotas quotas = new KafkaUserQuotas();
         quotas.setConsumerByteRate(2000);
-        JsonObject updated = new JsonObject(new String(kuq.updateUserJson(original.encode().getBytes(Charset.defaultCharset()), quotas), Charset.defaultCharset()));
+        JsonObject updated = new JsonObject(new String(kuq.updateUserJson(original.encode().getBytes(StandardCharsets.UTF_8), quotas), StandardCharsets.UTF_8));
         assertThat(updated.getJsonObject("config").getString("consumer_byte_rate"), is(notNullValue()));
         assertThat(updated.getJsonObject("config").getString("consumer_byte_rate"), is("2000"));
 
         quotas.setConsumerByteRate(3000);
         quotas.setProducerByteRate(4000);
         original = new JsonObject().put("version", 1).put("config", new JsonObject().put("consumer_byte_rate", "1000").put("producer_byte_rate", "2000"));
-        updated = new JsonObject(new String(kuq.updateUserJson(original.encode().getBytes(Charset.defaultCharset()), quotas), Charset.defaultCharset()));
+        updated = new JsonObject(new String(kuq.updateUserJson(original.encode().getBytes(StandardCharsets.UTF_8), quotas), StandardCharsets.UTF_8));
         assertThat(updated.getJsonObject("config").getString("consumer_byte_rate"), is(notNullValue()));
         assertThat(updated.getJsonObject("config").getString("producer_byte_rate"), is(notNullValue()));
         assertThat(updated.getJsonObject("config").getString("producer_byte_rate"), is("4000"));
@@ -181,7 +186,7 @@ public class KafkaUserQuotasIT {
         original = new JsonObject().put("version", 1).put("config", new JsonObject());
         quotas.setProducerByteRate(null);
         quotas.setConsumerByteRate(1000);
-        updated = new JsonObject(new String(kuq.updateUserJson(original.encode().getBytes(Charset.defaultCharset()), quotas), Charset.defaultCharset()));
+        updated = new JsonObject(new String(kuq.updateUserJson(original.encode().getBytes(StandardCharsets.UTF_8), quotas), StandardCharsets.UTF_8));
         assertThat(updated.getJsonObject("config").getString("consumer_byte_rate"), is(notNullValue()));
     }
 }
diff --git a/user-operator/src/test/java/io/strimzi/operator/user/operator/ScramShaCredentialsTest.java b/user-operator/src/test/java/io/strimzi/operator/user/operator/ScramShaCredentialsIT.java
similarity index 83%
rename from user-operator/src/test/java/io/strimzi/operator/user/operator/ScramShaCredentialsTest.java
rename to user-operator/src/test/java/io/strimzi/operator/user/operator/ScramShaCredentialsIT.java
index e174b614a..6b04446d7 100644
--- a/user-operator/src/test/java/io/strimzi/operator/user/operator/ScramShaCredentialsTest.java
+++ b/user-operator/src/test/java/io/strimzi/operator/user/operator/ScramShaCredentialsIT.java
@@ -12,7 +12,7 @@ import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
-import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
 
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.CoreMatchers.notNullValue;
@@ -21,7 +21,7 @@ import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.fail;
 
 
-public class ScramShaCredentialsTest {
+public class ScramShaCredentialsIT {
 
     private static EmbeddedZooKeeper zkServer;
 
@@ -57,6 +57,7 @@ public class ScramShaCredentialsTest {
     public void normalDelete() {
         scramShaCred.createOrUpdate("normalDelete", "foo-password");
         scramShaCred.delete("normalDelete");
+        assertThat(scramShaCred.isPathExist("/config/users/normalDelete"), is(false));
     }
 
     @Test
@@ -64,6 +65,7 @@ public class ScramShaCredentialsTest {
         scramShaCred.createOrUpdate("doubleDelete", "foo-password");
         scramShaCred.delete("doubleDelete");
         scramShaCred.delete("doubleDelete");
+        assertThat(scramShaCred.isPathExist("/config/users/doubleDelete"), is(false));
     }
 
     @Test
@@ -76,7 +78,6 @@ public class ScramShaCredentialsTest {
     public void userExists() {
         scramShaCred.createOrUpdate("userExists", "foo-password");
         assertThat(scramShaCred.exists("userExists"), is(true));
-
     }
 
     @Test
@@ -124,20 +125,22 @@ public class ScramShaCredentialsTest {
     @Test
     public void testDeletion()  {
         JsonObject original = new JsonObject().put("version", 1).put("config", new JsonObject().put("SCRAM-SHA-512", "somecredentials"));
-        JsonObject updated = new JsonObject(new String(scramShaCred.deleteUserJson(original.encode().getBytes(Charset.defaultCharset())), Charset.defaultCharset()));
+        original.getJsonObject("config").put("persist", 42);
+        JsonObject updated = scramShaCred.deleteUserJson(original.encode().getBytes(StandardCharsets.UTF_8));
         assertThat(updated.getJsonObject("config").getString("SCRAM-SHA-512"), is(nullValue()));
+        assertThat(updated.getJsonObject("config").getInteger("persist"), is(42));
 
         original = new JsonObject().put("version", 1).put("config", new JsonObject().put("SCRAM-SHA-512", "somecredentials").put("SCRAM-SHA-256", "somecredentials"));
-        updated = new JsonObject(new String(scramShaCred.deleteUserJson(original.encode().getBytes(Charset.defaultCharset())), Charset.defaultCharset()));
+        updated = scramShaCred.deleteUserJson(original.encode().getBytes(StandardCharsets.UTF_8));
         assertThat(updated.getJsonObject("config").getString("SCRAM-SHA-512"), is(nullValue()));
         assertThat(updated.getJsonObject("config").getString("SCRAM-SHA-256"), is("somecredentials"));
 
         original = new JsonObject().put("version", 1).put("config", new JsonObject());
-        updated = new JsonObject(new String(scramShaCred.deleteUserJson(original.encode().getBytes(Charset.defaultCharset())), Charset.defaultCharset()));
+        updated = scramShaCred.deleteUserJson(original.encode().getBytes(StandardCharsets.UTF_8));
         assertThat(updated.getJsonObject("config").getString("SCRAM-SHA-512"), is(nullValue()));
 
         original = new JsonObject().put("version", 1).put("config", new JsonObject().put("SCRAM-SHA-256", "somecredentials"));
-        updated = new JsonObject(new String(scramShaCred.deleteUserJson(original.encode().getBytes(Charset.defaultCharset())), Charset.defaultCharset()));
+        updated = scramShaCred.deleteUserJson(original.encode().getBytes(StandardCharsets.UTF_8));
         assertThat(updated.getJsonObject("config").getString("SCRAM-SHA-512"), is(nullValue()));
         assertThat(updated.getJsonObject("config").getString("SCRAM-SHA-256"), is("somecredentials"));
     }
@@ -145,20 +148,20 @@ public class ScramShaCredentialsTest {
     @Test
     public void testUpdate()  {
         JsonObject original = new JsonObject().put("version", 1).put("config", new JsonObject().put("SCRAM-SHA-512", "somecredentials"));
-        JsonObject updated = new JsonObject(new String(scramShaCred.updateUserJson(original.encode().getBytes(Charset.defaultCharset()), "password"), Charset.defaultCharset()));
+        JsonObject updated = new JsonObject(new String(scramShaCred.updateUserJson(original.encode().getBytes(StandardCharsets.UTF_8), "password"), StandardCharsets.UTF_8));
         assertThat(updated.getJsonObject("config").getString("SCRAM-SHA-512"), is(notNullValue()));
 
         original = new JsonObject().put("version", 1).put("config", new JsonObject().put("SCRAM-SHA-512", "somecredentials").put("SCRAM-SHA-256", "somecredentials"));
-        updated = new JsonObject(new String(scramShaCred.updateUserJson(original.encode().getBytes(Charset.defaultCharset()), "password"), Charset.defaultCharset()));
+        updated = new JsonObject(new String(scramShaCred.updateUserJson(original.encode().getBytes(StandardCharsets.UTF_8), "password"), StandardCharsets.UTF_8));
         assertThat(updated.getJsonObject("config").getString("SCRAM-SHA-512"), is(notNullValue()));
         assertThat(updated.getJsonObject("config").getString("SCRAM-SHA-256"), is("somecredentials"));
 
         original = new JsonObject().put("version", 1).put("config", new JsonObject());
-        updated = new JsonObject(new String(scramShaCred.updateUserJson(original.encode().getBytes(Charset.defaultCharset()), "password"), Charset.defaultCharset()));
+        updated = new JsonObject(new String(scramShaCred.updateUserJson(original.encode().getBytes(StandardCharsets.UTF_8), "password"), StandardCharsets.UTF_8));
         assertThat(updated.getJsonObject("config").getString("SCRAM-SHA-512"), is(notNullValue()));
 
         original = new JsonObject().put("version", 1).put("config", new JsonObject().put("SCRAM-SHA-256", "somecredentials"));
-        updated = new JsonObject(new String(scramShaCred.updateUserJson(original.encode().getBytes(Charset.defaultCharset()), "password"), Charset.defaultCharset()));
+        updated = new JsonObject(new String(scramShaCred.updateUserJson(original.encode().getBytes(StandardCharsets.UTF_8), "password"), StandardCharsets.UTF_8));
         assertThat(updated.getJsonObject("config").getString("SCRAM-SHA-512"), is(notNullValue()));
         assertThat(updated.getJsonObject("config").getString("SCRAM-SHA-256"), is("somecredentials"));
     }
['systemtest/src/test/java/io/strimzi/systemtest/UserST.java', 'user-operator/src/test/java/io/strimzi/operator/user/operator/ScramShaCredentialsTest.java', 'user-operator/src/main/java/io/strimzi/operator/user/operator/ScramShaCredentials.java', 'user-operator/src/test/java/io/strimzi/operator/user/operator/KafkaUserQuotasIT.java', 'user-operator/src/main/java/io/strimzi/operator/user/operator/KafkaUserQuotasOperator.java']