1 /*

     2  * Copyright 2007-2008 Sun Microsystems, Inc.  All Rights Reserved.

     3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

     4  *

     5  * This code is free software; you can redistribute it and/or modify it

     6  * under the terms of the GNU General Public License version 2 only, as

     7  * published by the Free Software Foundation.

     8  *

     9  * This code is distributed in the hope that it will be useful, but WITHOUT

    10  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

    11  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

    12  * version 2 for more details (a copy is included in the LICENSE file that

    13  * accompanied this code).

    14  *

    15  * You should have received a copy of the GNU General Public License version

    16  * 2 along with this work; if not, write to the Free Software Foundation,

    17  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

    18  *

    19  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,

    20  * CA 95054 USA or visit www.sun.com if you need additional information or

    21  * have any questions.

    22  */

    23 

    24 /*

    25  * @test ReconnectableJMXConnector

    26  * @bug 5108776

    27  * @summary Check that the Event Service can be used to build a

    28  * ReconnectableJMXConnector.

    29  * @author Eamonn McManus

    30  */

    31 

    32 import java.io.IOException;

    33 import java.lang.reflect.InvocationHandler;

    34 import java.lang.reflect.InvocationTargetException;

    35 import java.lang.reflect.Method;

    36 import java.lang.reflect.Proxy;

    37 import java.util.Date;

    38 import java.util.Map;

    39 import java.util.NoSuchElementException;

    40 import java.util.concurrent.ArrayBlockingQueue;

    41 import java.util.concurrent.BlockingQueue;

    42 import java.util.concurrent.TimeUnit;

    43 import java.util.concurrent.atomic.AtomicLong;

    44 import java.util.concurrent.atomic.AtomicReference;

    45 import java.util.concurrent.locks.Condition;

    46 import java.util.concurrent.locks.Lock;

    47 import java.util.concurrent.locks.ReentrantLock;

    48 import javax.management.ListenerNotFoundException;

    49 import javax.management.MBeanServer;

    50 import javax.management.MBeanServerConnection;

    51 import javax.management.MBeanServerFactory;

    52 import javax.management.Notification;

    53 import javax.management.NotificationBroadcasterSupport;

    54 import javax.management.NotificationFilter;

    55 import javax.management.NotificationListener;

    56 import javax.management.ObjectName;

    57 import javax.management.event.EventClient;

    58 import javax.management.remote.JMXConnectionNotification;

    59 import javax.management.remote.JMXConnector;

    60 import javax.management.remote.JMXConnectorFactory;

    61 import javax.management.remote.JMXConnectorServer;

    62 import javax.management.remote.JMXConnectorServerFactory;

    63 import javax.management.remote.JMXServiceURL;

    64 import javax.security.auth.Subject;

    65 

    66 /*

    67  * This test checks that it is possible to use the Event Service to create

    68  * a "reconnectable connector".

    69  *

    70  * In the JMX Remote API, we deliberately specified that a connector client

    71  * (JMXConnector) that encounters a network failure is then permanently broken.

    72  * The idea being that adding recovery logic to the basic connector client

    73  * would make it much more complicated and less reliable, and the logic would

    74  * in any case never correspond to what a given situation needs. Some of

    75  * the tough questions are: Should the connector try to mask the failure by

    76  * blocking operations until the failure is resolved? How long should the

    77  * connector try to reestablish the connection before giving up? Rather than

    78  * try to solve this problem in the connector, we suggested that people who

    79  * wanted to recover from network failures could implement the JMXConnector

    80  * interface themselves so that it forwards to a wrapped JMXConnector that can

    81  * be replaced in case of network failure.

    82  *

    83  * This works fine except that the connector client has state,

    84  * in the form of listeners added by the user through the

    85  * MBeanServerConnection.addNotificationListener method. It's possible

    86  * for the wrapper to keep track of these listeners as well as forwarding

    87  * them to the wrapped JMXConnector, so that it can reapply them to

    88  * a replacement JMXConnector after failure recover. But it's quite

    89  * tricky, particularly because of the two- and four-argument versions of

    90  * removeNotificationListener.

    91  *

    92  * The Event Service can take care of this for you through the EventClient

    93  * class. Listeners added through that class are implemented in a way that

    94  * doesn't require the connector client to maintain any state, so they should

    95  * continue to work transparently after replacing the wrapped JMXConnector.

    96  * This test is a proof of concept that shows it works.  Quite a number of

    97  * details would need to be changed to build a reliable reconnectable

    98  * connector.

    99  *

   100  * The test simulates network failure by rewrapping the wrapped JMXConnector's

   101  * MBeanServerConnection (MBSC) in a "breakable" MBSC which we can cause

   102  * to stop working.  We do this in two phases.  The first phase suspends

   103  * any MBSC calls just at the point where they would return to the caller.

   104  * The goal here is to block an EventClientDelegateMBean.fetchNotifications

   105  * operation when it has received notifications but not yet delivered them

   106  * to the EventClient.  This is the most delicate point where a breakage

   107  * can occur, because the EventClientDelegate must not drop those notifs

   108  * from its buffer until another fetchNotifs call arrives with a later

   109  * sequence number (which is an implicit ack of the previous set of

   110  * notifs).  Once the fetchNotifs call is suspended, we "kill" the MBSC,

   111  * causing it to throw IOException from this and any other calls.  That

   112  * triggers the reconnect logic, which will make a new MBSC and issue

   113  * the same fetchNotifs call to it.

   114  *

   115  * The test could be improved by synchronizing explicitly between the

   116  * breakable MBSC and the mainline, so we only proceed to kill the MBSC

   117  * when we are sure that the fetchNotifs call is blocked.  As it is,

   118  * we have a small delay which both ensures that no notifs are delivered

   119  * while the connection is suspended, and if the machine is fast enough

   120  * allows the fetchNotifs call to reach the blocking point.

   121  */

   122 public class ReconnectableConnectorTest {

   123     private static class ReconnectableJMXConnector implements JMXConnector {

   124         private final JMXServiceURL url;

   125         private AtomicReference<JMXConnector> wrappedJMXC =

   126                 new AtomicReference<JMXConnector>();

   127         private AtomicReference<MBeanServerConnection> wrappedMBSC =

   128                 new AtomicReference<MBeanServerConnection>();

   129         private final NotificationBroadcasterSupport broadcaster =

   130                 new NotificationBroadcasterSupport();

   131         private final Lock connectLock = new ReentrantLock();

   132 

   133         ReconnectableJMXConnector(JMXServiceURL url) {

   134             this.url = url;

   135         }

   136 

   137         private class ReconnectIH implements InvocationHandler {

   138             public Object invoke(Object proxy, Method method, Object[] args)

   139                     throws Throwable {

   140                 try {

   141                     return method.invoke(wrappedMBSC.get(), args);

   142                 } catch (InvocationTargetException e) {

   143                     if (e.getCause() instanceof IOException) {

   144                         connect();

   145                         try {

   146                             return method.invoke(wrappedMBSC.get(),args);

   147                         } catch (InvocationTargetException ee) {

   148                             throw ee.getCause();

   149                         }

   150                     }

   151                     throw e.getCause();

   152                 }

   153             }

   154         }

   155 

   156         private class FailureListener implements NotificationListener {

   157             public void handleNotification(Notification n, Object h) {

   158                 String type = n.getType();

   159                 if (type.equals(JMXConnectionNotification.FAILED)) {

   160                     try {

   161                         connect();

   162                     } catch (IOException e) {

   163                         broadcaster.sendNotification(n);

   164                     }

   165                 } else if (type.equals(JMXConnectionNotification.NOTIFS_LOST))

   166                     broadcaster.sendNotification(n);

   167             }

   168         }

   169 

   170         public void connect() throws IOException {

   171             connectLock.lock();

   172             try {

   173                 connectWithLock();

   174             } finally {

   175                 connectLock.unlock();

   176             }

   177         }

   178 

   179         private void connectWithLock() throws IOException {

   180             MBeanServerConnection mbsc = wrappedMBSC.get();

   181             if (mbsc != null) {

   182                 try {

   183                     mbsc.getDefaultDomain();

   184                     return;  // the connection works

   185                 } catch (IOException e) {

   186                     // OK: the connection doesn't work, so make a new one

   187                 }

   188             }

   189             // This is where we would need to add the fancy logic that

   190             // allows the connection to keep failing for a while

   191             // before giving up.

   192             JMXConnector jmxc = JMXConnectorFactory.connect(url);

   193             jmxc.addConnectionNotificationListener(

   194                     new FailureListener(), null, null);

   195             wrappedJMXC.set(jmxc);

   196             if (false)

   197                 wrappedMBSC.set(jmxc.getMBeanServerConnection());

   198             else {

   199                 mbsc = jmxc.getMBeanServerConnection();

   200                 InvocationHandler ih = new BreakableIH(mbsc);

   201                 mbsc = (MBeanServerConnection) Proxy.newProxyInstance(

   202                         MBeanServerConnection.class.getClassLoader(),

   203                         new Class<?>[] {MBeanServerConnection.class},

   204                         ih);

   205                 wrappedMBSC.set(mbsc);

   206             }

   207         }

   208 

   209         private BreakableIH breakableIH() {

   210             MBeanServerConnection mbsc = wrappedMBSC.get();

   211             return (BreakableIH) Proxy.getInvocationHandler(mbsc);

   212         }

   213 

   214         void suspend() {

   215             BreakableIH ih = breakableIH();

   216             ih.suspend();

   217         }

   218 

   219         void kill() throws IOException {

   220             BreakableIH ih = breakableIH();

   221             wrappedJMXC.get().close();

   222             ih.kill();

   223         }

   224 

   225         public void connect(Map<String, ?> env) throws IOException {

   226             throw new UnsupportedOperationException("Not supported yet.");

   227         }

   228 

   229         private final AtomicReference<MBeanServerConnection> mbscRef =

   230                 new AtomicReference<MBeanServerConnection>();

   231 

   232         public MBeanServerConnection getMBeanServerConnection()

   233                 throws IOException {

   234             connect();

   235             // Synchro here is not strictly correct: two threads could make

   236             // an MBSC at the same time.  OK for a test but beware for real

   237             // code.

   238             MBeanServerConnection mbsc = mbscRef.get();

   239             if (mbsc != null)

   240                 return mbsc;

   241             mbsc = (MBeanServerConnection) Proxy.newProxyInstance(

   242                     MBeanServerConnection.class.getClassLoader(),

   243                     new Class<?>[] {MBeanServerConnection.class},

   244                     new ReconnectIH());

   245             mbsc = EventClient.getEventClientConnection(mbsc);

   246             mbscRef.set(mbsc);

   247             return mbsc;

   248         }

   249 

   250         public MBeanServerConnection getMBeanServerConnection(

   251                 Subject delegationSubject) throws IOException {

   252             throw new UnsupportedOperationException("Not supported yet.");

   253         }

   254 

   255         public void close() throws IOException {

   256             wrappedJMXC.get().close();

   257         }

   258 

   259         public void addConnectionNotificationListener(

   260                 NotificationListener l, NotificationFilter f, Object h) {

   261             broadcaster.addNotificationListener(l, f, h);

   262         }

   263 

   264         public void removeConnectionNotificationListener(NotificationListener l)

   265                 throws ListenerNotFoundException {

   266             broadcaster.removeNotificationListener(l);

   267         }

   268 

   269         public void removeConnectionNotificationListener(

   270                 NotificationListener l, NotificationFilter f, Object h)

   271                 throws ListenerNotFoundException {

   272             broadcaster.removeNotificationListener(l, f, h);

   273         }

   274 

   275         public String getConnectionId() throws IOException {

   276             return wrappedJMXC.get().getConnectionId();

   277         }

   278     }

   279 

   280     // InvocationHandler that allows us to perform a two-phase "break" of

   281     // an object.  The first phase suspends the object, so that calls to

   282     // it are blocked just before they return.  The second phase unblocks

   283     // suspended threads and causes them to throw IOException.

   284     private static class BreakableIH implements InvocationHandler {

   285         private final Object wrapped;

   286         private final Holder<String> state = new Holder<String>("running");

   287 

   288         BreakableIH(Object wrapped) {

   289             this.wrapped = wrapped;

   290         }

   291 

   292         void suspend() {

   293             state.set("suspended");

   294         }

   295 

   296         void kill() {

   297             state.set("killed");

   298         }

   299 

   300         public Object invoke(Object proxy, Method method, Object[] args)

   301                 throws Throwable {

   302             Object result;

   303             try {

   304                 result = method.invoke(wrapped, args);

   305             } catch (InvocationTargetException e) {

   306                 throw e.getCause();

   307             }

   308             String s = state.get();

   309             if (s.equals("suspended"))

   310                 state.waitUntilEqual("killed", 3, TimeUnit.SECONDS);

   311             else if (s.equals("killed"))

   312                 throw new IOException("Broken");

   313             return result;

   314         }

   315     }

   316 

   317     private static class Holder<T> {

   318         private T held;

   319         private Lock lock = new ReentrantLock();

   320         private Condition changed = lock.newCondition();

   321 

   322         Holder(T value) {

   323             lock.lock();

   324             this.held = value;

   325             lock.unlock();

   326         }

   327 

   328         void waitUntilEqual(T value, long timeout, TimeUnit units)

   329                 throws InterruptedException {

   330             long millis = units.toMillis(timeout);

   331             long stop = System.currentTimeMillis() + millis;

   332             Date stopDate = new Date(stop);

   333             lock.lock();

   334             try {

   335                 while (!value.equals(held)) {

   336                     boolean ok = changed.awaitUntil(stopDate);

   337                     if (!ok)

   338                         throw new InterruptedException("Timed out");

   339                 }

   340             } finally {

   341                 lock.unlock();

   342             }

   343         }

   344 

   345         void set(T value) {

   346             lock.lock();

   347             try {

   348                 held = value;

   349                 changed.signalAll();

   350             } finally {

   351                 lock.unlock();

   352             }

   353         }

   354 

   355         T get() {

   356             lock.lock();

   357             try {

   358                 return held;

   359             } finally {

   360                 lock.unlock();

   361             }

   362         }

   363     }

   364 

   365     private static class StoreListener implements NotificationListener {

   366         final BlockingQueue<Notification> queue =

   367                 new ArrayBlockingQueue<Notification>(100);

   368 

   369         public void handleNotification(Notification n, Object h) {

   370             queue.add(n);

   371         }

   372 

   373         Notification nextNotification(long time, TimeUnit units)

   374                 throws InterruptedException {

   375             Notification n = queue.poll(time, units);

   376             if (n == null)

   377                 throw new NoSuchElementException("Notification wait timed out");

   378             return n;

   379         }

   380 

   381         int notifCount() {

   382             return queue.size();

   383         }

   384     }

   385 

   386     public static interface SenderMBean {}

   387     public static class Sender

   388             extends NotificationBroadcasterSupport implements SenderMBean {

   389         private AtomicLong seqNo = new AtomicLong(0);

   390 

   391         void send() {

   392             Notification n =

   393                     new Notification("type", this, seqNo.getAndIncrement());

   394             sendNotification(n);

   395         }

   396     }

   397 

   398     public static void main(String[] args) throws Exception {

   399         MBeanServer mbs = MBeanServerFactory.newMBeanServer();

   400         Sender sender = new Sender();

   401         ObjectName name = new ObjectName("a:b=c");

   402         mbs.registerMBean(sender, name);

   403 

   404         System.out.println("Creating connector server");

   405         JMXServiceURL url = new JMXServiceURL("service:jmx:rmi:///");

   406         JMXConnectorServer cs = JMXConnectorServerFactory.newJMXConnectorServer(

   407                 url, null, mbs);

   408         cs.start();

   409 

   410         StoreListener csListener = new StoreListener();

   411         cs.addNotificationListener(csListener, null, null);

   412 

   413         System.out.println("Creating reconnectable client");

   414         JMXServiceURL addr = cs.getAddress();

   415         ReconnectableJMXConnector cc = new ReconnectableJMXConnector(addr);

   416         MBeanServerConnection mbsc = cc.getMBeanServerConnection();

   417 

   418         System.out.println("Checking server has sent new-client notif");

   419         Notification csn = csListener.nextNotification(1, TimeUnit.SECONDS);

   420         assertEquals("CS notif type",

   421                 JMXConnectionNotification.OPENED, csn.getType());

   422 

   423         StoreListener listener = new StoreListener();

   424         mbsc.addNotificationListener(name, listener, null, null);

   425 

   426         System.out.println("Sending 10 notifs and checking they are received");

   427         for (int i = 0; i < 10; i++)

   428             sender.send();

   429         checkNotifs(listener, 0, 10);

   430 

   431         System.out.println("Suspending the fetchNotifs operation");

   432         cc.suspend();

   433         System.out.println("Sending a notif while fetchNotifs is suspended");

   434         sender.send();

   435         System.out.println("Brief wait before checking no notif is received");

   436         Thread.sleep(2);

   437         // dumpThreads();

   438         assertEquals("notif queue while connector suspended",

   439                 0, listener.notifCount());

   440         assertEquals("connector server notif queue while connector suspended",

   441                 0, csListener.notifCount());

   442 

   443         System.out.println("Breaking the connection so fetchNotifs will fail over");

   444         cc.kill();

   445 

   446         System.out.println("Checking that client has reconnected");

   447         csn = csListener.nextNotification(1, TimeUnit.SECONDS);

   448         assertEquals("First CS notif type after kill",

   449                 JMXConnectionNotification.CLOSED, csn.getType());

   450         csn = csListener.nextNotification(1, TimeUnit.SECONDS);

   451         assertEquals("Second CS notif type after kill",

   452                 JMXConnectionNotification.OPENED, csn.getType());

   453 

   454         System.out.println("Checking that suspended notif has been received");

   455         checkNotifs(listener, 10, 11);

   456     }

   457 

   458     private static void checkNotifs(

   459              StoreListener sl, long start, long stop)

   460             throws Exception {

   461         for (long i = start; i < stop; i++) {

   462             Notification n = sl.nextNotification(1, TimeUnit.SECONDS);

   463             assertEquals("received sequence number", i, n.getSequenceNumber());

   464         }

   465     }

   466 

   467     private static void assertEquals(String what, Object expect, Object actual)

   468     throws Exception {

   469         if (!expect.equals(actual)) {

   470             fail(what + " should be " + expect + " but is " + actual);

   471         }

   472     }

   473 

   474     private static void fail(String why) throws Exception {

   475         throw new Exception("TEST FAILED: " + why);

   476     }

   477 

   478     private static void dumpThreads() {

   479         System.out.println("Thread stack dump");

   480         Map<Thread, StackTraceElement[]> traces = Thread.getAllStackTraces();

   481         for (Map.Entry<Thread, StackTraceElement[]> entry : traces.entrySet()) {

   482             Thread t = entry.getKey();

   483             System.out.println("===Thread " + t.getName() + "===");

   484             for (StackTraceElement ste : entry.getValue())

   485                 System.out.println("    " + ste);

   486         }

   487     }

   488 }