Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-4357: LeaderInit: Add failed to lock event

Browse Source 
JIRA: https://jira.spring.io/browse/INT-4357

Updated based on code review.

Updated adocs

*Polishing code style, typos
* Add protection from NPE when `LockRegistryLeaderInitiator` isn't
supplied with the `leaderEventPublisher`
* Add `What's New` note
This commit is contained in:
Glenn Renfro
2017-10-12 15:50:59 -04:00
committed by Artem Bilan
parent a7eebed252
commit 1979f91cf5
9 changed files with 186 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
spring-integration-core/src/main/java/org/springframework/integration/leader/event/DefaultLeaderEventPublisher.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2016 the original author or authors.
* Copyright 2015-2017 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -25,6 +25,7 @@ import org.springframework.integration.leader.Context;
*
* @author Janne Valkealahti
* @author Gary Russell
* @author Glenn Renfro
*
*/
public class DefaultLeaderEventPublisher implements LeaderEventPublisher, ApplicationEventPublisherAware {
@@ -60,6 +61,13 @@ public class DefaultLeaderEventPublisher implements LeaderEventPublisher, Applic
}
}
@Override
public void publishOnFailedToAcquire(Object source, Context context, String role) {
if (this.applicationEventPublisher != null) {
this.applicationEventPublisher.publishEvent(new OnFailedToAcquireMutexEvent(source, context, role));
}
}
@Override
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
this.applicationEventPublisher = applicationEventPublisher;

14
spring-integration-core/src/main/java/org/springframework/integration/leader/event/LeaderEventPublisher.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2016 the original author or authors.
* Copyright 2015-2017 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -23,13 +23,13 @@ import org.springframework.integration.leader.Context;
*
* @author Janne Valkealahti
* @author Gary Russell
* @author Glenn Renfro
*
*/
public interface LeaderEventPublisher {
/**
* Publish a granted event.
*
* @param source the component generated this event
* @param context the context associated with event
* @param role the role of the leader
@@ -38,11 +38,19 @@ public interface LeaderEventPublisher {
/**
* Publish a revoked event.
*
* @param source the component generated this event
* @param context the context associated with event
* @param role the role of the leader
*/
void publishOnRevoked(Object source, Context context, String role);
/**
* Publish a failure to acquire event.
* @param source the component generated this event
* @param context the context associated with event
* @param role the role of the leader
* @since 5.0
*/
void publishOnFailedToAcquire(Object source, Context context, String role);
}

41
spring-integration-core/src/main/java/org/springframework/integration/leader/event/OnFailedToAcquireMutexEvent.java Normal file
View File

@@ -0,0 +1,41 @@
/*
* Copyright 2017 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.springframework.integration.leader.event;
import org.springframework.integration.leader.Context;
/**
* Generic event representing that a mutex could not be acquired during leader election.
*
* @author Glenn Renfro
* @author Artem Bilan
*
* @since 5.0
*/
public class OnFailedToAcquireMutexEvent extends AbstractLeaderEvent {
/**
* Instantiate a new OnFailedToAcquireMutexEvent.
* @param source the component that published the event (never {@code null})
* @param context the context associated with this event
* @param role the role of the leader
*/
public OnFailedToAcquireMutexEvent(Object source, Context context, String role) {
super(source, context, role);
}
}

39
spring-integration-core/src/main/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiator.java
View File

@@ -56,6 +56,8 @@ import org.springframework.util.Assert;
* @author Dave Syer
* @author Artem Bilan
* @author Vedran Pavic
* @author Glenn Renfro
*
* @since 4.3.1
*/
public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBean, ApplicationEventPublisherAware {
@@ -111,6 +113,8 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
*/
private long busyWaitMillis = DEFAULT_BUSY_WAIT_TIME;
private boolean publishFailedEvents = false;
private LeaderSelector leaderSelector;
private ApplicationEventPublisher applicationEventPublisher;
@@ -231,6 +235,24 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
return this.leaderSelector.context;
}
public boolean isPublishFailedEvents() {
return this.publishFailedEvents;
}
/**
* Enable or disable the publishing of failed events to the
* specified applicationEventPublisher. Because of the large
* number of failure events that can be published while attempting to get a
* mutex during leader election (in the case that another instance is
* holding the mutex), the default is set to false.
* @param publishFailedEvents boolean that if true, failed events will
* be published. If false, no failures will be published. Default is false.
* @since 5.0
*/
public void setPublishFailedEvents(boolean publishFailedEvents) {
this.publishFailedEvents = publishFailedEvents;
}
/**
* Start the registration of the {@link #candidate} for leader election.
*/
@@ -315,6 +337,9 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
this.locked = true;
handleGranted();
}
else if (isPublishFailedEvents()) {
publishFailedToAcquire();
}
}
else if (acquired) {
// If we were able to acquire it but we were already locked we
@@ -387,6 +412,20 @@ public class LockRegistryLeaderInitiator implements SmartLifecycle, DisposableBe
}
}
private void publishFailedToAcquire() {
if (LockRegistryLeaderInitiator.this.leaderEventPublisher != null) {
try {
LockRegistryLeaderInitiator.this.leaderEventPublisher.publishOnFailedToAcquire(
LockRegistryLeaderInitiator.this,
this.context,
LockRegistryLeaderInitiator.this.candidate.getRole());
}
catch (Exception e) {
logger.warn("Error publishing OnFailedToAcquire event.", e);
}
}
}
}
/**

48
spring-integration-core/src/test/java/org/springframework/integration/support/leader/LockRegistryLeaderInitiatorTests.java
View File

@@ -53,6 +53,7 @@ import org.springframework.integration.support.locks.LockRegistry;
* @author Dave Syer
* @author Artem Bilan
* @author Vedran Pavic
* @author Glenn Renfro
*
* @since 4.3.1
*/
@@ -115,6 +116,25 @@ public class LockRegistryLeaderInitiatorTests {
another.stop();
}
@Test
public void competingWithErrorPublish() throws Exception {
LockRegistryLeaderInitiator another =
new LockRegistryLeaderInitiator(this.registry, new DefaultCandidate());
CountDownLatch other = new CountDownLatch(1);
CountDownLatch failedAcquireLatch = new CountDownLatch(1);
another.setLeaderEventPublisher(new CountingPublisher(other, new CountDownLatch(1), failedAcquireLatch));
another.setPublishFailedEvents(true);
this.initiator.start();
assertThat(this.granted.await(20, TimeUnit.SECONDS), is(true));
another.start();
assertThat(failedAcquireLatch.await(20, TimeUnit.SECONDS), is(true));
this.initiator.stop();
assertThat(other.await(20, TimeUnit.SECONDS), is(true));
assertThat(another.getContext().isLeader(), is(true));
another.stop();
}
@Test
public void testExceptionFromEvent() throws Exception {
CountDownLatch onGranted = new CountDownLatch(1);
@@ -159,9 +179,10 @@ public class LockRegistryLeaderInitiatorTests {
new LockRegistryLeaderInitiator(firstRegistry, new DefaultCandidate());
CountDownLatch firstGranted = new CountDownLatch(1);
CountDownLatch firstRevoked = new CountDownLatch(1);
CountDownLatch firstAquireLockFailed = new CountDownLatch(1);
first.setHeartBeatMillis(10);
first.setBusyWaitMillis(1);
first.setLeaderEventPublisher(new CountingPublisher(firstGranted, firstRevoked));
first.setLeaderEventPublisher(new CountingPublisher(firstGranted, firstRevoked, firstAquireLockFailed));
// set up second registry instance - this one will NOT be able to obtain lock initially
LockRegistry secondRegistry = mock(LockRegistry.class);
@@ -174,9 +195,10 @@ public class LockRegistryLeaderInitiatorTests {
new LockRegistryLeaderInitiator(secondRegistry, new DefaultCandidate());
CountDownLatch secondGranted = new CountDownLatch(1);
CountDownLatch secondRevoked = new CountDownLatch(1);
CountDownLatch secondAquireLockFailed = new CountDownLatch(1);
second.setHeartBeatMillis(10);
second.setBusyWaitMillis(1);
second.setLeaderEventPublisher(new CountingPublisher(secondGranted, secondRevoked));
second.setLeaderEventPublisher(new CountingPublisher(secondGranted, secondRevoked, secondAquireLockFailed));
// start initiators
first.start();
@@ -249,13 +271,20 @@ public class LockRegistryLeaderInitiatorTests {
private final CountDownLatch revoked;
CountingPublisher(CountDownLatch granted, CountDownLatch revoked) {
this.granted = granted;
this.revoked = revoked;
}
private final CountDownLatch acquireFailed;
CountingPublisher(CountDownLatch granted) {
this(granted, new CountDownLatch(1));
this(granted, new CountDownLatch(1), new CountDownLatch(1));
}
CountingPublisher(CountDownLatch granted, CountDownLatch revoked) {
this(granted, revoked, new CountDownLatch(1));
}
CountingPublisher(CountDownLatch granted, CountDownLatch revoked, CountDownLatch acquireFailed) {
this.granted = granted;
this.revoked = revoked;
this.acquireFailed = acquireFailed;
}
@Override
@@ -263,6 +292,11 @@ public class LockRegistryLeaderInitiatorTests {
this.revoked.countDown();
}
@Override
public void publishOnFailedToAcquire(Object source, Context context, String role) {
this.acquireFailed.countDown();
}
@Override
public void publishOnGranted(Object source, Context context, String role) {
this.granted.countDown();

20
spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/leader/JdbcLockRegistryLeaderInitiatorTests.java
View File

@@ -45,6 +45,7 @@ import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType;
/**
* @author Artem Bilan
* @author Gary Russell
* @author Glenn Renfro
*
* @since 4.3.1
*/
@@ -110,7 +111,10 @@ public class JdbcLockRegistryLeaderInitiatorTests {
final CountDownLatch granted2 = new CountDownLatch(1);
CountDownLatch revoked1 = new CountDownLatch(1);
CountDownLatch revoked2 = new CountDownLatch(1);
initiator1.setLeaderEventPublisher(new CountingPublisher(granted1, revoked1) {
final CountDownLatch acquireLockFailed1 = new CountDownLatch(1);
final CountDownLatch acquireLockFailed2 = new CountDownLatch(1);
initiator1.setLeaderEventPublisher(new CountingPublisher(granted1, revoked1, acquireLockFailed1) {
@Override
public void publishOnRevoked(Object source, Context context, String role) {
@@ -126,7 +130,7 @@ public class JdbcLockRegistryLeaderInitiatorTests {
});
initiator2.setLeaderEventPublisher(new CountingPublisher(granted2, revoked2) {
initiator2.setLeaderEventPublisher(new CountingPublisher(granted2, revoked2, acquireLockFailed2) {
@Override
public void publishOnRevoked(Object source, Context context, String role) {
@@ -176,15 +180,18 @@ public class JdbcLockRegistryLeaderInitiatorTests {
private final CountDownLatch revoked;
private final CountDownLatch acquireLockFailed;
private volatile LockRegistryLeaderInitiator initiator;
CountingPublisher(CountDownLatch granted, CountDownLatch revoked) {
CountingPublisher(CountDownLatch granted, CountDownLatch revoked, CountDownLatch acquireLockFailed) {
this.granted = granted;
this.revoked = revoked;
this.acquireLockFailed = acquireLockFailed;
}
CountingPublisher(CountDownLatch granted) {
this(granted, new CountDownLatch(1));
this(granted, new CountDownLatch(1), new CountDownLatch(1));
}
@Override
@@ -192,6 +199,11 @@ public class JdbcLockRegistryLeaderInitiatorTests {
this.revoked.countDown();
}
@Override
public void publishOnFailedToAcquire(Object source, Context context, String role) {
this.acquireLockFailed.countDown();
}
@Override
public void publishOnGranted(Object source, Context context, String role) {
this.initiator = (LockRegistryLeaderInitiator) source;

22
spring-integration-redis/src/test/java/org/springframework/integration/redis/leader/RedisLockRegistryLeaderInitiatorTests.java
View File

@@ -38,6 +38,7 @@ import org.springframework.integration.support.leader.LockRegistryLeaderInitiato
/**
* @author Artem Bilan
* @author Gary Russell
* @author Glenn Renfro
*
* @since 4.3.9
*/
@@ -83,7 +84,10 @@ public class RedisLockRegistryLeaderInitiatorTests extends RedisAvailableTests {
final CountDownLatch granted2 = new CountDownLatch(1);
CountDownLatch revoked1 = new CountDownLatch(1);
CountDownLatch revoked2 = new CountDownLatch(1);
initiator1.setLeaderEventPublisher(new CountingPublisher(granted1, revoked1) {
CountDownLatch acquireLockFailed1 = new CountDownLatch(1);
CountDownLatch acquireLockFailed2 = new CountDownLatch(1);
initiator1.setLeaderEventPublisher(new CountingPublisher(granted1, revoked1, acquireLockFailed1) {
@Override
public void publishOnRevoked(Object source, Context context, String role) {
@@ -99,7 +103,7 @@ public class RedisLockRegistryLeaderInitiatorTests extends RedisAvailableTests {
});
initiator2.setLeaderEventPublisher(new CountingPublisher(granted2, revoked2) {
initiator2.setLeaderEventPublisher(new CountingPublisher(granted2, revoked2, acquireLockFailed2) {
@Override
public void publishOnRevoked(Object source, Context context, String role) {
@@ -132,7 +136,7 @@ public class RedisLockRegistryLeaderInitiatorTests extends RedisAvailableTests {
initiator2.stop();
CountDownLatch revoked11 = new CountDownLatch(1);
initiator1.setLeaderEventPublisher(new CountingPublisher(new CountDownLatch(1), revoked11));
initiator1.setLeaderEventPublisher(new CountingPublisher(new CountDownLatch(1), revoked11, new CountDownLatch(1)));
initiator1.getContext().yield();
@@ -150,13 +154,16 @@ public class RedisLockRegistryLeaderInitiatorTests extends RedisAvailableTests {
private volatile LockRegistryLeaderInitiator initiator;
CountingPublisher(CountDownLatch granted, CountDownLatch revoked) {
private final CountDownLatch acquireLockFailed;
CountingPublisher(CountDownLatch granted, CountDownLatch revoked, CountDownLatch acquireLockFailed) {
this.granted = granted;
this.revoked = revoked;
this.acquireLockFailed = acquireLockFailed;
}
CountingPublisher(CountDownLatch granted) {
this(granted, new CountDownLatch(1));
this(granted, new CountDownLatch(1), new CountDownLatch(1));
}
@Override
@@ -164,6 +171,11 @@ public class RedisLockRegistryLeaderInitiatorTests extends RedisAvailableTests {
this.revoked.countDown();
}
@Override
public void publishOnFailedToAcquire(Object source, Context context, String role) {
this.acquireLockFailed.countDown();
}
@Override
public void publishOnGranted(Object source, Context context, String role) {
this.initiator = (LockRegistryLeaderInitiator) source;

12
src/reference/asciidoc/endpoint.adoc
View File

@@ -695,12 +695,14 @@ This is useful in clustered scenarios where shared resources must only be consum
An example of this is a file inbound channel adapter that is polling a shared directory.
(See <<file-reading>>).
To participate in a leader election and be notified when elected leader or when leadership is revoked, an application creates a component in the application context called a "leader initiator".
To participate in a leader election and be notified when elected leader, when leadership is revoked or, failure to acquire the resources to become leader, an application creates a component in the application context called a "leader initiator".
Normally a leader initiator is a `SmartLifecycle` so it starts up (optionally) automatically when the context starts, and then publishes notifications when leadership changes.
Users can also receive failure notifications by setting the `publishFailedEvents` to `true` (starting with _version 5.0_), in cases when they want take a specific action if a failure occurs.
By convention the user provides a `Candidate` that receives the callbacks and also can revoke the leadership through a `Context` object provided by the framework.
User code can also listen for `AbstractLeaderEvents`, and respond accordingly, for instance using a `SmartLifecycleRoleController`.
User code can also listen for `org.springframework.integration.leader.event.AbstractLeaderEvent` s, and respond accordingly, for instance using a `SmartLifecycleRoleController`.
There is a basic implementation of a leader initiator based on the `LockRegistry` abstraction. To use it you just need to create an instance as a bean, for example:
There is a basic implementation of a leader initiator based on the `LockRegistry` abstraction.
To use it you just need to create an instance as a bean, for example:
[source, java]
----
@@ -710,6 +712,8 @@ public LockRegistryLeaderInitiator leaderInitiator(LockRegistry locks) {
}
----
If the lock registry is implemented correctly, there will only ever be at most one leader. If the lock registry also provides locks which throw exceptions (ideally `InterruptedException`) when they expire or are broken, then the duration of the leaderless periods can be as short as is allowed by the inherent latency in the lock implementation. By default there is a `busyWaitMillis` property that adds some additional latency to prevent CPU starvation in the (more usual) case that the locks are imperfect and you only know they expired by trying to obtain one again.
If the lock registry is implemented correctly, there will only ever be at most one leader.
If the lock registry also provides locks which throw exceptions (ideally `InterruptedException`) when they expire or are broken, then the duration of the leaderless periods can be as short as is allowed by the inherent latency in the lock implementation.
By default there is a `busyWaitMillis` property that adds some additional latency to prevent CPU starvation in the (more usual) case that the locks are imperfect and you only know they expired by trying to obtain one again.
See <<zk-leadership>> for more information about leadership election and events using Zookeeper.

4
src/reference/asciidoc/whats-new.adoc
View File

@@ -99,6 +99,10 @@ The `@GlobalChannelInterceptor` annotation and `<int:channel-interceptor>` now s
See <<global-channel-configuration-interceptors>> for more information.
A new `OnFailedToAcquireMutexEvent` is emitted now via `DefaultLeaderEventPublisher` by the `LockRegistryLeaderInitiator`, when candidate is failed to acquire the lock.
See <<leadership-event-handling>> for more information.
==== Gateway Changes
The gateway now correctly sets the `errorChannel` header when the gateway method has a `void` return type and an error channel is provided.