[controller] Fixed a bug that logger is trying to access a removed ma…

…p entry (#1298)

The top store admin operation entry is already removed before the logger trying
to access the top entry. This PR fixes this bug by storing the offset of
top entry before hand.

services/venice-controller/src/main/java/com/linkedin/venice/controller/kafka/consumer/AdminConsumptionTask.java

@@ -503,7 +503,8 @@ private void executeMessagesAndCollectResults() throws InterruptedException {
     boolean skipOffsetCommandHasBeenProcessed = false;
     for (Map.Entry<String, Queue<AdminOperationWrapper>> entry: storeAdminOperationsMapWithOffset.entrySet()) {
       if (!entry.getValue().isEmpty()) {
-        if (checkOffsetToSkip(entry.getValue().peek().getOffset(), false)) {
+        long adminMessageOffset = entry.getValue().peek().getOffset();
+        if (checkOffsetToSkip(adminMessageOffset, false)) {
           entry.getValue().remove();
           skipOffsetCommandHasBeenProcessed = true;
         }
@@ -526,7 +527,7 @@ private void executeMessagesAndCollectResults() throws InterruptedException {
           LOGGER.info(
               "Adding admin message from store {} with offset {} to the task list",
               entry.getKey(),
-              entry.getValue().peek().getOffset());
+              adminMessageOffset);
           this.tasks.add(newTask);
           stores.add(entry.getKey());
         }