[grid] Rejecting events when a Secret cannot be parsed.

This might happen if Grid is deployed to an environment
like Kubernetes, where checks are usually done on every
open port. This leads to unexpected payloads to be sent
to the EventBus ports, which then cannot be properly
parsed and then the listeners threads are interrupted.

This change tries to parse the Secret and if not possible,
the message is rejected. Also, the thread won't be
interrupted anymore.

Complements solution pushed for #9959

java/src/org/openqa/selenium/events/zeromq/UnboundZmqEventBus.java

@@ -17,9 +17,13 @@
 
 package org.openqa.selenium.events.zeromq;
 
+import static java.nio.charset.StandardCharsets.UTF_8;
+
 import com.google.common.collect.EvictingQueue;
+
 import net.jodah.failsafe.Failsafe;
 import net.jodah.failsafe.RetryPolicy;
+
 import org.openqa.selenium.events.Event;
 import org.openqa.selenium.events.EventBus;
 import org.openqa.selenium.events.EventListener;
@@ -54,8 +58,6 @@
 import java.util.logging.Level;
 import java.util.logging.Logger;
 
-import static java.nio.charset.StandardCharsets.UTF_8;
-
 class UnboundZmqEventBus implements EventBus {
 
   static final EventName REJECTED_EVENT = new EventName("selenium-rejected-event");
@@ -226,8 +228,16 @@ public void run() {
               ZMQ.Socket socket = poller.getSocket(i);
 
               EventName eventName = new EventName(new String(socket.recv(), UTF_8));
-              Secret eventSecret =
-                JSON.toType(new String(socket.recv(), UTF_8), Secret.class);
+
+              Secret eventSecret;
+              String receivedEventSecret = new String(socket.recv(), UTF_8);
+              try {
+                eventSecret = JSON.toType(receivedEventSecret, Secret.class);
+              } catch (JsonException e) {
+                rejectEvent(eventName, receivedEventSecret);
+                return;
+              }
+
               UUID id = UUID.fromString(new String(socket.recv(), UTF_8));
               String data = new String(socket.recv(), UTF_8);
 
@@ -242,33 +252,36 @@ public void run() {
               recentMessages.add(id);
 
               if (!Secret.matches(secret, eventSecret)) {
-                LOG.log(Level.SEVERE, "Received message without a valid secret. Rejecting. {0} -> {1}",
-                  new Object[]{event, data}); // String formatting only applied if needed
-                Event rejectedEvent =
-                  new Event(REJECTED_EVENT, new ZeroMqEventBus.RejectedEvent(eventName, data));
-
-                notifyListeners(REJECTED_EVENT, rejectedEvent);
-
+                rejectEvent(eventName, data);
                 return;
               }
+
               notifyListeners(eventName, event);
             }
           }
         } catch (Exception e) {
           if (e.getCause() instanceof AssertionError) {
             // Do nothing.
-          } else if (e instanceof JsonException) {
-            LOG.log(Level.WARNING, e, () -> "Caught exception while parsing for event bus messages: "
-                                            + e.getMessage());
           } else {
-            LOG.log(Level.WARNING, e, () -> "Caught exception while polling for event bus messages: "
-              + e.getMessage());
-            throw e;
+            LOG.log(Level.WARNING, e,
+                    () -> "Caught exception while polling for event bus messages: " +
+                          e.getMessage());
           }
         }
       }
     }
 
+    private void rejectEvent(EventName eventName, String data) {
+      Event rejectedEvent = new Event(REJECTED_EVENT,
+                                      new ZeroMqEventBus.RejectedEvent(eventName, data));
+      LOG.log(Level.SEVERE,
+              "Received message without a valid secret. Rejecting. {0} -> {1}",
+              new Object[]{rejectedEvent, data}); // String formatting only applied if needed
+
+      notifyListeners(REJECTED_EVENT, rejectedEvent);
+    }
+
+
     private void notifyListeners(EventName eventName, Event event) {
       List<Consumer<Event>> eventListeners = listeners.getOrDefault(eventName, new ArrayList<>());
       eventListeners