Make the LearnerContextImpl use the CappedLogger instead of CappedOpe…

…rations
neo4j · Dec 30, 2015 · 5e7ca5f · 5e7ca5f
1 parent cb58e4e
commit 5e7ca5f
Show file tree

Hide file tree

Showing 9 changed files with 86 additions and 70 deletions.
diff --git a/community/kernel/src/main/java/org/neo4j/kernel/impl/util/CappedLogger.java b/community/kernel/src/main/java/org/neo4j/kernel/impl/util/CappedLogger.java
@@ -46,6 +46,10 @@ public class CappedLogger
 
     public CappedLogger( StringLogger delegate )
     {
+        if ( delegate == null )
+        {
+            throw new IllegalArgumentException( "The delegate StringLogger cannot be null" );
+        }
         filter = new Filter();
         this.delegate = delegate;
     }
@@ -98,21 +102,23 @@ public void reset()
      * Set a limit to the amount of logging that this logger will accept between resets.
      * @param limit The number of log messages that the CappedLogger will let through in between resets.
      */
-    public void setCountLimit( int limit )
+    public CappedLogger setCountLimit( int limit )
     {
         if ( limit < 1 )
         {
             throw new IllegalArgumentException( "The count limit must be positive" );
         }
         filter = filter.setCountLimit( limit );
+        return this;
     }
 
     /**
      * Unset the count limit, and allow any number of messages through, provided other limits don't apply.
      */
-    public void unsetCountLimit()
+    public CappedLogger unsetCountLimit()
     {
         filter = filter.unsetCountLimit();
+        return this;
     }
 
     /**
@@ -122,7 +128,7 @@ public void unsetCountLimit()
      * @param unit The time unit.
      * @param clock The clock to use for reading the current time when checking this limit.
      */
-    public void setTimeLimit( long time, TimeUnit unit, Clock clock )
+    public CappedLogger setTimeLimit( long time, TimeUnit unit, Clock clock )
     {
         if ( time < 1 )
         {
@@ -137,15 +143,17 @@ public void setTimeLimit( long time, TimeUnit unit, Clock clock )
             throw new IllegalArgumentException( "The clock used for time limiting cannot be null" );
         }
         filter = filter.setTimeLimit( time, unit, clock );
+        return this;
     }
 
     /**
      * Unset the time limit, and allow any number of messages through, as often as possible, provided other limits
      * don't apply.
      */
-    public void unsetTimeLimit()
+    public CappedLogger unsetTimeLimit()
     {
         filter = filter.unsetTimeLimit();
+        return this;
     }
 
     /**
@@ -154,9 +162,10 @@ public void unsetTimeLimit()
      * messages will get logged in full.
      * @param enabled {@code true} if duplicates should be filtered, {@code false} if they should not.
      */
-    public void setDuplicateFilterEnabled( boolean enabled )
+    public CappedLogger setDuplicateFilterEnabled( boolean enabled )
     {
         filter = filter.setDuplicateFilterEnabled( enabled );
+        return this;
     }
 
     private static class Filter

diff --git a/community/kernel/src/test/java/org/neo4j/kernel/impl/util/CappedLoggerTest.java b/community/kernel/src/test/java/org/neo4j/kernel/impl/util/CappedLoggerTest.java
@@ -171,6 +171,12 @@ public void setUp()
         logger = new CappedLogger( delegate );
     }
 
+    @Test( expected = IllegalArgumentException.class )
+    public void mustThrowIfDelegateIsNull() throws Exception
+    {
+        new CappedLogger( null );
+    }
+
     @Test
     public void mustLogWithoutLimitConfiguration() throws Exception
     {
@@ -225,6 +231,17 @@ public void mustThrowOnNullClock() throws Exception
         logger.setTimeLimit( 10, TimeUnit.MILLISECONDS, null );
     }
 
+    @Test
+    public void mustAllowConfigurationChaining() throws Exception
+    {
+        logger.setCountLimit( 1 )
+              .setTimeLimit( 10, TimeUnit.MILLISECONDS, Clock.SYSTEM_CLOCK )
+              .setDuplicateFilterEnabled( true )
+              .unsetCountLimit()
+              .unsetTimeLimit()
+              .setCountLimit( 1 );
+    }
+
     @Test
     public void mustLimitByConfiguredCount() throws Exception
     {

diff --git a/...ava/org/neo4j/cluster/protocol/atomicbroadcast/multipaxos/context/LearnerContextImpl.java b/...ava/org/neo4j/cluster/protocol/atomicbroadcast/multipaxos/context/LearnerContextImpl.java
@@ -30,7 +30,7 @@
 import org.neo4j.cluster.protocol.atomicbroadcast.multipaxos.PaxosInstanceStore;
 import org.neo4j.cluster.protocol.heartbeat.HeartbeatContext;
 import org.neo4j.cluster.timeout.Timeouts;
-import org.neo4j.kernel.impl.util.CappedOperation;
+import org.neo4j.kernel.impl.util.CappedLogger;
 import org.neo4j.kernel.logging.Logging;
 
 class LearnerContextImpl
@@ -42,19 +42,7 @@ class LearnerContextImpl
     private long lastLearnedInstanceId = -1;
 
     /** To minimize logging, keep track of the latest learn miss, only log when it changes. */
-    private final CappedOperation<org.neo4j.cluster.protocol.atomicbroadcast.multipaxos.InstanceId> learnMissLogging =
+    private final CappedLogger learnMissLogger;
-            new CappedOperation<org.neo4j.cluster.protocol.atomicbroadcast.multipaxos.InstanceId>(
-                    CappedOperation.differentItems() )
-            {
-                @Override
-                protected void triggered( InstanceId instanceId )
-                {
-                    getConsoleLogger( LearnerState.class ).warn( "Did not have learned value for Paxos instance "
-                            + instanceId + ". This generally indicates that this instance has missed too many " +
-                            "cluster events and is failing to catch up. If this error does not resolve soon it " +
-                            "may become necessary to restart this cluster member so normal operation can resume.");
-                }
-            };
 
     private final HeartbeatContext heartbeatContext;
     private final AcceptorInstanceStore instanceStore;
@@ -76,6 +64,7 @@ protected void triggered( InstanceId instanceId )
         this.objectInputStreamFactory = objectInputStreamFactory;
         this.objectOutputStreamFactory = objectOutputStreamFactory;
         this.paxosInstances = paxosInstances;
+        this.learnMissLogger = new CappedLogger( getLogger( LearnerState.class ) ).setDuplicateFilterEnabled( true );
     }
 
     private LearnerContextImpl( org.neo4j.cluster.InstanceId me, CommonContextState commonState, Logging logging,
@@ -92,6 +81,7 @@ private LearnerContextImpl( org.neo4j.cluster.InstanceId me, CommonContextState
         this.objectInputStreamFactory = objectInputStreamFactory;
         this.objectOutputStreamFactory = objectOutputStreamFactory;
         this.paxosInstances = paxosInstances;
+        this.learnMissLogger = new CappedLogger( getLogger( LearnerState.class ) ).setDuplicateFilterEnabled( true );
     }
 
     @Override
@@ -183,7 +173,10 @@ public void setNextInstanceId( long id )
     @Override
     public void notifyLearnMiss( InstanceId instanceId )
     {
-        learnMissLogging.event( instanceId );
+        learnMissLogger.warn( "Did not have learned value for Paxos instance " + instanceId + ". " +
+                              "This generally indicates that this instance has missed too many cluster events and is " +
+                              "failing to catch up. If this error does not resolve soon it may become necessary to " +
+                              "restart this cluster member so normal operation can resume.", null );
     }
 
     public LearnerContextImpl snapshot( CommonContextState commonStateSnapshot, Logging logging, Timeouts timeouts,

diff --git a/...c/test/java/org/neo4j/cluster/protocol/atomicbroadcast/multipaxos/LearnerContextTest.java b/...c/test/java/org/neo4j/cluster/protocol/atomicbroadcast/multipaxos/LearnerContextTest.java
@@ -26,33 +26,37 @@
 import org.neo4j.cluster.protocol.cluster.ClusterConfiguration;
 import org.neo4j.cluster.protocol.election.ElectionRole;
 import org.neo4j.helpers.collection.Iterables;
+import org.neo4j.kernel.logging.DevNullLoggingService;
+import org.neo4j.kernel.logging.Logging;
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.equalTo;
 import static org.mockito.Mockito.mock;
 
 public class LearnerContextTest
 {
+    private final Logging logging = new DevNullLoggingService();
+
     @Test
     public void shouldOnlyAllowHigherLastLearnedInstanceId() throws Exception
     {
         // Given
-        MultiPaxosContext mpCtx = new MultiPaxosContext( null, Iterables.<ElectionRole>empty(), mock( ClusterConfiguration.class ), null, null, null, null, null, null, null );
+        MultiPaxosContext mpCtx = new MultiPaxosContext( null, Iterables.<ElectionRole>empty(), mock( ClusterConfiguration.class ), null, logging, null, null, null, null, null );
         LearnerContext state = mpCtx.getLearnerContext();
 
         // When
         state.setLastKnownLearnedInstanceInCluster( 1, new InstanceId( 2 ) );
         state.setLastKnownLearnedInstanceInCluster( 0, new InstanceId( 3 ) );
 
         // Then
-        assertThat( state.getLastKnownLearnedInstanceInCluster(), equalTo( 1l ) );
+        assertThat( state.getLastKnownLearnedInstanceInCluster(), equalTo( 1L ) );
     }
 
     @Test
     public void shouldTrackLastKnownUpToDateAliveInstance() throws Exception
     {
         // Given
-        MultiPaxosContext mpCtx = new MultiPaxosContext( null, Iterables.<ElectionRole>empty(), mock( ClusterConfiguration.class ), null, null, null, null, null, null, null );
+        MultiPaxosContext mpCtx = new MultiPaxosContext( null, Iterables.<ElectionRole>empty(), mock( ClusterConfiguration.class ), null, logging, null, null, null, null, null );
         LearnerContext state = mpCtx.getLearnerContext();
 
         // When
@@ -61,23 +65,23 @@ public void shouldTrackLastKnownUpToDateAliveInstance() throws Exception
         state.setLastKnownLearnedInstanceInCluster( 0, new InstanceId( 4 ) );
 
         // Then
-        assertThat( state.getLastKnownLearnedInstanceInCluster(), equalTo( 1l ) );
+        assertThat( state.getLastKnownLearnedInstanceInCluster(), equalTo( 1L ) );
         assertThat( state.getLastKnownAliveUpToDateInstance(), equalTo( new InstanceId( 3 ) ));
     }
 
     @Test
     public void settingLastLearnedInstanceToNegativeOneShouldAlwaysWin() throws Exception
     {
         // Given
-        MultiPaxosContext mpCtx = new MultiPaxosContext( null, Iterables.<ElectionRole>empty(), mock( ClusterConfiguration.class ), null, null, null, null, null, null, null );
+        MultiPaxosContext mpCtx = new MultiPaxosContext( null, Iterables.<ElectionRole>empty(), mock( ClusterConfiguration.class ), null, logging, null, null, null, null, null );
         LearnerContext state = mpCtx.getLearnerContext();
 
         // When
         state.setLastKnownLearnedInstanceInCluster( 1, new InstanceId( 2 ) );
         state.setLastKnownLearnedInstanceInCluster( -1, null );
 
         // Then
-        assertThat( state.getLastKnownLearnedInstanceInCluster(), equalTo( -1l ) );
+        assertThat( state.getLastKnownLearnedInstanceInCluster(), equalTo( -1L ) );
         assertThat( state.getLastKnownAliveUpToDateInstance(), equalTo( new org.neo4j.cluster.InstanceId( 2 ) ));
     }
 }
diff --git a/...org/neo4j/cluster/protocol/atomicbroadcast/multipaxos/context/LearnerContextImplTest.java b/...org/neo4j/cluster/protocol/atomicbroadcast/multipaxos/context/LearnerContextImplTest.java
@@ -20,23 +20,20 @@
 package org.neo4j.cluster.protocol.atomicbroadcast.multipaxos.context;
 
 import org.junit.Test;
-import org.mockito.Matchers;
-
-import java.util.ArrayList;
-import java.util.List;
 
 import org.neo4j.cluster.InstanceId;
 import org.neo4j.cluster.protocol.atomicbroadcast.ObjectInputStreamFactory;
 import org.neo4j.cluster.protocol.atomicbroadcast.ObjectOutputStreamFactory;
 import org.neo4j.cluster.protocol.atomicbroadcast.multipaxos.AcceptorInstanceStore;
 import org.neo4j.cluster.protocol.atomicbroadcast.multipaxos.PaxosInstanceStore;
 import org.neo4j.cluster.timeout.Timeouts;
-import org.neo4j.kernel.logging.ConsoleLogger;
+import org.neo4j.kernel.impl.util.StringLogger;
+import org.neo4j.kernel.logging.DevNullLoggingService;
 import org.neo4j.kernel.logging.Logging;
 
+import static org.hamcrest.Matchers.containsString;
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertThat;
-import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 
 public class LearnerContextImplTest
@@ -45,35 +42,34 @@ public class LearnerContextImplTest
     public void shouldOnlyLogLearnMissOnce() throws Exception
     {
         // Given
-        Logging logging = mock( Logging.class );
+        StringBuffer buffer = new StringBuffer();
-        LearnerContextImpl ctx = new LearnerContextImpl( new InstanceId( 1 ), mock( CommonContextState.class ),
+        final StringLogger logger = StringLogger.wrap( buffer );
-                logging, mock( Timeouts.class ), mock( PaxosInstanceStore.class ), mock( AcceptorInstanceStore.class ),
+        Logging logging = new DevNullLoggingService()
-                mock( ObjectInputStreamFactory.class ), mock( ObjectOutputStreamFactory.class ),
-                mock( HeartbeatContextImpl.class ) );
-
-        final List<String> logs = new ArrayList<>();
-        ConsoleLogger consoleLogger = new ConsoleLogger( null )
         {
             @Override
-            public void warn( String message )
+            public StringLogger getMessagesLog( Class loggingClass )
             {
-                logs.add( message );
+                return logger;
             }
         };
-        doReturn( consoleLogger ).when( logging ).getConsoleLog( Matchers.<Class> any() );
+        LearnerContextImpl ctx = new LearnerContextImpl( new InstanceId( 1 ), mock( CommonContextState.class ),
+                logging, mock( Timeouts.class ), mock( PaxosInstanceStore.class ), mock( AcceptorInstanceStore.class ),
+                mock( ObjectInputStreamFactory.class ), mock( ObjectOutputStreamFactory.class ),
+                mock( HeartbeatContextImpl.class ) );
 
         // When
-        ctx.notifyLearnMiss( new org.neo4j.cluster.protocol.atomicbroadcast.multipaxos.InstanceId( 1l ) );
+        ctx.notifyLearnMiss( new org.neo4j.cluster.protocol.atomicbroadcast.multipaxos.InstanceId( 1L ) );
-        ctx.notifyLearnMiss( new org.neo4j.cluster.protocol.atomicbroadcast.multipaxos.InstanceId( 1l ) );
+        ctx.notifyLearnMiss( new org.neo4j.cluster.protocol.atomicbroadcast.multipaxos.InstanceId( 1L ) );
-        ctx.notifyLearnMiss( new org.neo4j.cluster.protocol.atomicbroadcast.multipaxos.InstanceId( 2l ) );
+        ctx.notifyLearnMiss( new org.neo4j.cluster.protocol.atomicbroadcast.multipaxos.InstanceId( 2L ) );
-        ctx.notifyLearnMiss( new org.neo4j.cluster.protocol.atomicbroadcast.multipaxos.InstanceId( 2l ) );
+        ctx.notifyLearnMiss( new org.neo4j.cluster.protocol.atomicbroadcast.multipaxos.InstanceId( 2L ) );
-        ctx.notifyLearnMiss( new org.neo4j.cluster.protocol.atomicbroadcast.multipaxos.InstanceId( 1l ) );
+        ctx.notifyLearnMiss( new org.neo4j.cluster.protocol.atomicbroadcast.multipaxos.InstanceId( 1L ) );
 
         // Then
-        assertEquals( 3, logs.size() );
+        String[] logs = buffer.toString().split( "\n" );
-        assertTrue( logs.get( 0 ).startsWith( "Did not have learned value for Paxos instance 1." ) );
+        assertEquals( 3, logs.length );
-        assertTrue( logs.get( 1 ).startsWith( "Did not have learned value for Paxos instance 2." ) );
+        assertThat( logs[0], containsString( "Did not have learned value for Paxos instance 1." ) );
-        assertTrue( logs.get( 2 ).startsWith( "Did not have learned value for Paxos instance 1." ) );
+        assertThat( logs[1], containsString( "Did not have learned value for Paxos instance 2." ) );
+        assertThat( logs[2], containsString( "Did not have learned value for Paxos instance 1." ) );
     }
 
 }
diff --git a/...rprise/cluster/src/test/java/org/neo4j/cluster/protocol/election/ElectionContextTest.java b/...rprise/cluster/src/test/java/org/neo4j/cluster/protocol/election/ElectionContextTest.java
@@ -19,6 +19,9 @@
  */
 package org.neo4j.cluster.protocol.election;
 
+import org.junit.Test;
+import org.mockito.Matchers;
+
 import java.net.URI;
 import java.util.Collections;
 import java.util.HashMap;
@@ -29,9 +32,6 @@
 import java.util.Set;
 import java.util.concurrent.Executor;
 
-import org.junit.Test;
-import org.mockito.Matchers;
-
 import org.neo4j.cluster.InstanceId;
 import org.neo4j.cluster.protocol.atomicbroadcast.ObjectInputStreamFactory;
 import org.neo4j.cluster.protocol.atomicbroadcast.ObjectOutputStreamFactory;
@@ -43,6 +43,7 @@
 import org.neo4j.cluster.timeout.Timeouts;
 import org.neo4j.helpers.collection.Iterables;
 import org.neo4j.kernel.impl.util.StringLogger;
+import org.neo4j.kernel.logging.DevNullLoggingService;
 import org.neo4j.kernel.logging.Logging;
 
 import static junit.framework.Assert.assertNull;
@@ -99,7 +100,7 @@ public void testElectionNotOkQuorumFailedTwoInstances()
 
         MultiPaxosContext context = new MultiPaxosContext( new InstanceId( 1 ),
                 Iterables.<ElectionRole, ElectionRole>iterable( new ElectionRole( "coordinator" ) ),
-                clusterConfiguration, mock( Executor.class ), mock( Logging.class ),
+                clusterConfiguration, mock( Executor.class ), new DevNullLoggingService(),
                 mock( ObjectInputStreamFactory.class ), mock( ObjectOutputStreamFactory.class ),
                 mock( AcceptorInstanceStore.class ), mock( Timeouts.class ),
                 mock( ElectionCredentialsProvider.class) );
@@ -132,7 +133,7 @@ public void testElectionNotOkQuorumFailedFourInstances()
 
         MultiPaxosContext context = new MultiPaxosContext( new InstanceId(1), Iterables.<ElectionRole, ElectionRole>iterable(
                 new ElectionRole( "coordinator" ) ), clusterConfiguration,
-                mock( Executor.class ), mock( Logging.class ),
+                mock( Executor.class ), new DevNullLoggingService(),
                 mock( ObjectInputStreamFactory.class ), mock( ObjectOutputStreamFactory.class ),
                 mock( AcceptorInstanceStore.class ), mock( Timeouts.class ),
                 mock( ElectionCredentialsProvider.class) );
@@ -167,7 +168,7 @@ public void testElectionNotOkQuorumFailedFiveInstances()
 
         MultiPaxosContext context = new MultiPaxosContext( new InstanceId(1), Iterables.<ElectionRole, ElectionRole>iterable(
                 new ElectionRole( "coordinator" ) ), clusterConfiguration,
-                mock( Executor.class ), mock( Logging.class ),
+                mock( Executor.class ), new DevNullLoggingService(),
                 mock( ObjectInputStreamFactory.class ), mock( ObjectOutputStreamFactory.class ),
                 mock( AcceptorInstanceStore.class ), mock( Timeouts.class ),
                 mock( ElectionCredentialsProvider.class) );
@@ -446,7 +447,7 @@ private void baseTestForElectionOk( Set<InstanceId> failed, boolean moreThanQuor
 
         MultiPaxosContext context = new MultiPaxosContext( new InstanceId(1), Iterables.<ElectionRole, ElectionRole>iterable(
                         new ElectionRole( "coordinator" ) ), clusterConfiguration,
-                        mock( Executor.class ), mock( Logging.class ),
+                        mock( Executor.class ), new DevNullLoggingService(),
                         mock( ObjectInputStreamFactory.class ), mock( ObjectOutputStreamFactory.class ),
                 mock( AcceptorInstanceStore.class ), mock( Timeouts.class ),
                 mock( ElectionCredentialsProvider.class) );

diff --git a/enterprise/ha/src/main/java/org/neo4j/kernel/ha/SlaveUpdatePuller.java b/enterprise/ha/src/main/java/org/neo4j/kernel/ha/SlaveUpdatePuller.java
@@ -166,10 +166,8 @@ public boolean evaluate( int currentTicket, int targetTicket )
         this.invalidEpochHandler = invalidEpochHandler;
         this.jobScheduler = jobScheduler;
         this.logger = logging.getMessagesLog( getClass() );
-        this.invalidEpochCappedLogger = new CappedLogger( logger );
+        this.invalidEpochCappedLogger = new CappedLogger( logger ).setCountLimit( LOG_CAP );
-        this.comExceptionCappedLogger = new CappedLogger( logger );
+        this.comExceptionCappedLogger = new CappedLogger( logger ).setCountLimit( LOG_CAP );
-        invalidEpochCappedLogger.setCountLimit( LOG_CAP );
-        comExceptionCappedLogger.setCountLimit( LOG_CAP );
     }
 
     @Override