Changes:

- nuked TrailLicense
 - Move license expiry enforcement logic to LicensesService
 - clean up ESLicenseManager
 - make notification scheduling logic as lazy as possible
 - make sure to notify from cluster changed only if needed
 - added tests for notification

Original commit: elastic/x-pack-elasticsearch@e31b682f41

src/main/java/org/elasticsearch/license/core/ESLicense.java

@@ -113,6 +113,7 @@ public class ESLicense implements Comparable<ESLicense> {
 
     @Override
     public int compareTo(ESLicense o) {
+        assert o != null;
         return Long.compare(expiryDate, o.expiryDate);
     }
 
@@ -269,44 +270,38 @@ public class ESLicense implements Comparable<ESLicense> {
                     .signature(license.signature());
         }
 
+        public ESLicense verifyAndBuild() {
+            verify();
+            return new ESLicense(uid, issuer, issuedTo, issueDate, type,
+                    subscriptionType, feature, signature, expiryDate, maxNodes);
+        }
+
         public ESLicense build() {
-            verify(false);
             return new ESLicense(uid, issuer, issuedTo, issueDate, type,
                     subscriptionType, feature, signature, expiryDate, maxNodes);
         }
 
-        public ESLicense buildInternal() {
+        private void verify() {
-            verify(true);
-            return new ESLicense(uid, issuer, issuedTo, issueDate, type,
-                    subscriptionType, feature, signature, expiryDate, maxNodes);
-        }
-
-        private void verify(boolean internal) {
-            String msg = null;
             if (issuer == null) {
-                msg = "issuer can not be null";
+               throw new IllegalStateException("issuer can not be null");
             } else if (issuedTo == null) {
-                msg = "issuedTo can not be null";
+               throw new IllegalStateException("issuedTo can not be null");
             } else if (issueDate == -1) {
-                msg = "issueDate has to be set";
+               throw new IllegalStateException("issueDate has to be set");
             } else if (type == null) {
-                msg = "type can not be null";
+               throw new IllegalStateException("type can not be null");
             } else if (subscriptionType == null) {
-                msg = "subscriptionType can not be null";
+               throw new IllegalStateException("subscriptionType can not be null");
             } else if (uid == null) {
-                msg = "uid can not be null";
+               throw new IllegalStateException("uid can not be null");
             } else if (feature == null) {
-                msg = "at least one feature has to be enabled";
+               throw new IllegalStateException("at least one feature has to be enabled");
-            } else if (internal && signature == null) {
+            } else if (signature == null) {
-                msg = "signature can not be null";
+               throw new IllegalStateException("signature can not be null");
             } else if (maxNodes == -1) {
-                msg = "maxNodes has to be set";
+               throw new IllegalStateException("maxNodes has to be set");
             } else if (expiryDate == -1) {
-                msg = "expiryDate has to be set";
+               throw new IllegalStateException("expiryDate has to be set");
-            }
-
-            if (msg != null) {
-                throw new IllegalStateException(msg);
             }
         }
     }
@@ -354,7 +349,7 @@ public class ESLicense implements Comparable<ESLicense> {
                 .issueDate((long) map.get(Fields.ISSUE_DATE))
                 .expiryDate((long) map.get(Fields.EXPIRY_DATE))
                 .issuer((String) map.get(Fields.ISSUER))
-                .build();
+                .verifyAndBuild();
     }
 
     static ESLicense readFrom(StreamInput in) throws IOException {
@@ -370,7 +365,7 @@ public class ESLicense implements Comparable<ESLicense> {
                 .issuedTo((String) licenseMap.get(Fields.ISSUED_TO))
                 .signature((String) licenseMap.get(Fields.SIGNATURE))
                 .issuer((String) licenseMap.get(Fields.ISSUER))
-                .build();
+                .verifyAndBuild();
     }
 
     static void writeTo(ESLicense esLicense, StreamOutput out) throws IOException {

src/main/java/org/elasticsearch/license/core/ESLicenses.java

@@ -5,6 +5,7 @@
  */
 package org.elasticsearch.license.core;
 
+import org.elasticsearch.common.collect.ImmutableMap;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.xcontent.XContentBuilder;
@@ -62,4 +63,24 @@ public class ESLicenses {
         }
 
     }
+
+    public static ImmutableMap<String, ESLicense> reduceAndMap(Set<ESLicense> esLicensesSet) {
+        Map<String, ESLicense> map = new HashMap<>(esLicensesSet.size());
+        for (ESLicense license : esLicensesSet) {
+            putIfAppropriate(map, license);
+        }
+        return ImmutableMap.copyOf(map);
+    }
+
+    private static void putIfAppropriate(Map<String, ESLicense> licenseMap, ESLicense license) {
+        final String featureType = license.feature();
+        if (licenseMap.containsKey(featureType)) {
+            final ESLicense previousLicense = licenseMap.get(featureType);
+            if (license.expiryDate() > previousLicense.expiryDate()) {
+                licenseMap.put(featureType, license);
+            }
+        } else if (license.expiryDate() > System.currentTimeMillis()) {
+            licenseMap.put(featureType, license);
+        }
+    }
 }

src/main/java/org/elasticsearch/license/licensor/ESLicenseSigner.java

@@ -127,7 +127,7 @@ public class ESLicenseSigner {
                 .expiryDate(licenseSpec.expiryDate)
                 .issuer(licenseSpec.issuer)
                 .signature(signature)
-                .build();
+                .verifyAndBuild();
     }
 
 }

src/main/java/org/elasticsearch/license/licensor/tools/FileBasedESLicenseProvider.java

@@ -7,7 +7,7 @@ package org.elasticsearch.license.licensor.tools;
 
 import org.elasticsearch.common.collect.ImmutableMap;
 import org.elasticsearch.license.core.ESLicense;
-import org.elasticsearch.license.manager.Utils;
+import org.elasticsearch.license.core.ESLicenses;
 
 import java.util.Map;
 import java.util.Set;
@@ -18,7 +18,7 @@ public class FileBasedESLicenseProvider {
     private ImmutableMap<String, ESLicense> esLicenses;
 
     public FileBasedESLicenseProvider(Set<ESLicense> esLicenses) {
-        this.esLicenses = Utils.reduceAndMap(esLicenses);
+        this.esLicenses = ESLicenses.reduceAndMap(esLicenses);
     }
 
     public ESLicense getESLicense(String feature) {
@@ -31,6 +31,6 @@ public class FileBasedESLicenseProvider {
 
     // For testing
     public void setLicenses(Set<ESLicense> esLicenses) {
-        this.esLicenses = Utils.reduceAndMap(esLicenses);
+        this.esLicenses = ESLicenses.reduceAndMap(esLicenses);
     }
 }

src/main/java/org/elasticsearch/license/manager/ESLicenseManager.java

@@ -6,7 +6,6 @@
 package org.elasticsearch.license.manager;
 
 import net.nicholaswilliams.java.licensing.*;
-import net.nicholaswilliams.java.licensing.encryption.FilePublicKeyDataProvider;
 import net.nicholaswilliams.java.licensing.encryption.Hasher;
 import net.nicholaswilliams.java.licensing.encryption.PasswordProvider;
 import net.nicholaswilliams.java.licensing.exception.ExpiredLicenseException;
@@ -29,8 +28,6 @@ import static org.elasticsearch.license.manager.Utils.extractSignedLicence;
  * Class responsible for reading signed licenses, maintaining an effective esLicenses instance, verification of licenses
  * and querying against licenses on a feature basis
  * <p/>
- * TODO:
- *  - make it into a guice singleton
  */
 public class ESLicenseManager {
 
@@ -61,10 +58,26 @@ public class ESLicenseManager {
         this.licenseManager = LicenseManager.getInstance();
     }
 
-    public void verifyLicenses(Map<String, org.elasticsearch.license.core.ESLicense> esLicenses) {
+    public ImmutableSet<String> toSignatures(Collection<ESLicense> esLicenses) {
+        Set<String> signatures = new HashSet<>();
+        for (ESLicense esLicense : esLicenses) {
+            signatures.add(esLicense.signature());
+        }
+        return ImmutableSet.copyOf(signatures);
+    }
+
+    public ImmutableSet<ESLicense> fromSignatures(Set<String> signatures) {
+        Set<ESLicense> esLicenses = new HashSet<>();
+        for (String signature : signatures) {
+            esLicenses.add(fromSignature(signature));
+        }
+        return ImmutableSet.copyOf(esLicenses);
+    }
+
+    public void verifyLicenses(Map<String, ESLicense> esLicenses) {
         try {
             for (String feature : esLicenses.keySet()) {
-                org.elasticsearch.license.core.ESLicense esLicense = esLicenses.get(feature);
+                ESLicense esLicense = esLicenses.get(feature);
                 // verify signature
                 final License license = this.licenseManager.decryptAndVerifyLicense(
                         extractSignedLicence(esLicense.signature()));
@@ -81,31 +94,11 @@ public class ESLicenseManager {
         }
     }
 
-    public ImmutableSet<String> toSignatures(Collection<ESLicense> esLicenses) {
+    private ESLicense fromSignature(String signature) {
-        Set<String> signatures = new HashSet<>();
-        for (ESLicense esLicense : esLicenses) {
-            signatures.add(esLicense.signature());
-        }
-        return ImmutableSet.copyOf(signatures);
-    }
-
-    public ImmutableSet<ESLicense> fromSignatures(Set<String> signatures) {
-        Set<ESLicense> esLicenses = new HashSet<>();
-
-        for (String signature : signatures) {
-            ESLicense license = fromSignature(signature);
-            esLicenses.add(license);
-        }
-        return ImmutableSet.copyOf(esLicenses);
-    }
-
-    public ESLicense fromSignature(String signature) {
         final SignedLicense signedLicense = Utils.extractSignedLicence(signature);
         License license = licenseManager.decryptAndVerifyLicense(signedLicense);
         ESLicense.Builder builder = ESLicense.builder();
 
-
-
         for (License.Feature feature : license.getFeatures()) {
             String featureName = feature.getName();
             if (featureName.startsWith(Prefix.MAX_NODES)) {
@@ -155,42 +148,10 @@ public class ESLicenseManager {
             }
         }
         if (!licenseValid || !featureValid || !maxNodesValid || !typeValid || !subscriptionTypeValid) {
-            //only for debugging
-            String msg = "licenseValid: " + licenseValid + "\n" +
-                    "featureValid: " + featureValid + "\n" +
-                    "maxNodeValid: " + maxNodesValid + "\n" +
-                    "typeValid: " + typeValid + "\n" +
-                    "subscriptionTypeValid: " + subscriptionTypeValid + "\n";
             throw new InvalidLicenseException("Invalid License");
         }
     }
 
-
-    public boolean hasLicenseForFeature(String feature, Map<String, ESLicense> licenseMap) {
-        try {
-            final License license = getInternalLicense(feature, licenseMap);
-            if (license != null) {
-                return license.hasLicenseForFeature("feature:" + feature);
-            }
-            return false;
-        } catch (ExpiredLicenseException e) {
-            return false;
-        } catch (InvalidLicenseException e) {
-            return false;
-        }
-    }
-
-    private License getInternalLicense(String feature,  Map<String, ESLicense> licenseMap) {
-        ESLicense esLicense = licenseMap.get(feature);
-        if (esLicense != null) {
-            String signature = esLicense.signature();
-            License license = this.licenseManager.decryptAndVerifyLicense(extractSignedLicence(signature));
-            this.licenseManager.validateLicense(license);
-            return license;
-        }
-        return null;
-    }
-
     // TODO: Need a better password management
     private static class ESPublicKeyPasswordProvider implements PasswordProvider {
         private final String DEFAULT_PASS_PHRASE = "elasticsearch-license";

src/main/java/org/elasticsearch/license/manager/Utils.java

@@ -36,26 +36,4 @@ public final class Utils {
         int version = byteBuffer.getInt();
         return new ObjectSerializer().readObject(SignedLicense.class, Arrays.copyOfRange(signatureBytes, start, signatureBytes.length));
     }
-
-
-    public static ImmutableMap<String, ESLicense> reduceAndMap(Set<ESLicense> esLicensesSet) {
-        Map<String, ESLicense> map = new HashMap<>(esLicensesSet.size());
-        for (ESLicense license : esLicensesSet) {
-            putIfAppropriate(map, license);
-        }
-        return ImmutableMap.copyOf(map);
-    }
-
-
-    private static void putIfAppropriate(Map<String, ESLicense> licenseMap, ESLicense license) {
-        final String featureType = license.feature();
-        if (licenseMap.containsKey(featureType)) {
-            final ESLicense previousLicense = licenseMap.get(featureType);
-            if (license.expiryDate() > previousLicense.expiryDate()) {
-                licenseMap.put(featureType, license);
-            }
-        } else if (license.expiryDate() > System.currentTimeMillis()) {
-            licenseMap.put(featureType, license);
-        }
-    }
 }

src/main/java/org/elasticsearch/license/plugin/action/get/GetLicenseResponse.java

@@ -13,9 +13,7 @@ import org.elasticsearch.license.core.ESLicenses;
 
 import java.io.IOException;
 import java.util.ArrayList;
-import java.util.HashSet;
 import java.util.List;
-import java.util.Set;
 
 public class GetLicenseResponse extends ActionResponse {
 

src/main/java/org/elasticsearch/license/plugin/action/get/TransportGetLicenseAction.java

@@ -21,7 +21,6 @@ import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.TransportService;
 
 import java.util.List;
-import java.util.Set;
 
 public class TransportGetLicenseAction extends TransportMasterNodeReadOperationAction<GetLicenseRequest, GetLicenseResponse> {
 

src/main/java/org/elasticsearch/license/plugin/action/put/PutLicenseRequestBuilder.java

@@ -8,12 +8,9 @@ package org.elasticsearch.license.plugin.action.put;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.master.AcknowledgedRequestBuilder;
 import org.elasticsearch.client.ClusterAdminClient;
-import org.elasticsearch.common.collect.Lists;
 import org.elasticsearch.license.core.ESLicense;
 
-import java.util.ArrayList;
 import java.util.List;
-import java.util.Set;
 
 /**
  * Register license request builder

src/main/java/org/elasticsearch/license/plugin/core/LicensesMetaData.java

@@ -57,10 +57,13 @@ public class LicensesMetaData implements MetaData.Custom {
 
     @Override
     public boolean equals(Object obj) {
+        if (obj == null) {
+            return false;
+        }
         if (obj instanceof LicensesMetaData) {
             LicensesMetaData other = (LicensesMetaData) obj;
-            boolean signaturesEqual = false;
+            boolean signaturesEqual;
-            boolean trialLicensesEqual = false;
+            boolean trialLicensesEqual;
 
             if (other.getSignatures() != null) {
                 if (this.getSignatures() != null) {
@@ -144,12 +147,12 @@ public class LicensesMetaData implements MetaData.Custom {
                 }
                 if (fieldName != null) {
                     if (fieldName.equals(Fields.LICENSES)) {
-                        while ((token = parser.nextToken()) != XContentParser.Token.END_ARRAY) {
+                        while (parser.nextToken() != XContentParser.Token.END_ARRAY) {
                             signatures.add(parser.text());
                         }
                     }
                     if (fieldName.equals(Fields.TRIAL_LICENSES)) {
-                        while ((token = parser.nextToken()) != XContentParser.Token.END_ARRAY) {
+                        while (parser.nextToken() != XContentParser.Token.END_ARRAY) {
                             encodedTrialLicenses.add(parser.text());
                         }
                     }

src/main/java/org/elasticsearch/license/plugin/core/LicensesService.java

@@ -30,9 +30,6 @@ import org.elasticsearch.license.core.ESLicense;
 import org.elasticsearch.license.manager.ESLicenseManager;
 import org.elasticsearch.license.plugin.action.delete.DeleteLicenseRequest;
 import org.elasticsearch.license.plugin.action.put.PutLicenseRequest;
-import org.elasticsearch.license.plugin.core.trial.TrialLicenseUtils;
-import org.elasticsearch.license.plugin.core.trial.TrialLicenses;
-import org.elasticsearch.license.plugin.core.trial.TrialLicensesBuilder;
 import org.elasticsearch.threadpool.ThreadPool;
 import org.elasticsearch.transport.*;
 
@@ -43,9 +40,9 @@ import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.ScheduledFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 
-import static org.elasticsearch.license.manager.Utils.reduceAndMap;
+import static org.elasticsearch.license.core.ESLicenses.reduceAndMap;
-import static org.elasticsearch.license.plugin.core.trial.TrialLicenses.TrialLicense;
 
 /**
  * Service responsible for managing {@link org.elasticsearch.license.plugin.core.LicensesMetaData}
@@ -73,7 +70,9 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
 
     private Queue<ListenerHolder> pendingRegistrations = new ConcurrentLinkedQueue<>();
 
-    private volatile ScheduledFuture notificationScheduler;
+    private final AtomicReference<ScheduledFuture> notificationScheduler;
+
+    private final AtomicReference<LicensesMetaData> lastObservedState;
 
     @Inject
     public LicensesService(Settings settings, ClusterService clusterService, ThreadPool threadPool, TransportService transportService, ESLicenseManager esLicenseManager) {
@@ -82,6 +81,8 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
         this.esLicenseManager = esLicenseManager;
         this.threadPool = threadPool;
         this.transportService = transportService;
+        this.lastObservedState = new AtomicReference<>(null);
+        this.notificationScheduler = new AtomicReference<>(null);
         transportService.registerHandler(REGISTER_TRIAL_LICENSE_ACTION_NAME, new RegisterTrialLicenseRequestHandler());
     }
 
@@ -117,8 +118,8 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
                 MetaData.Builder mdBuilder = MetaData.builder(currentState.metaData());
                 LicensesMetaData currentLicenses = metaData.custom(LicensesMetaData.TYPE);
                 final LicensesWrapper licensesWrapper = LicensesWrapper.wrap(currentLicenses);
-                licensesWrapper.addSignedLicenses(esLicenseManager, Sets.newHashSet(newLicenses));
+                licensesWrapper.addSignedLicenses(esLicenseManager, newLicenses);
-                mdBuilder.putCustom(LicensesMetaData.TYPE, licensesWrapper.createLicensesMetaData());
+                mdBuilder.putCustom(LicensesMetaData.TYPE, licensesWrapper.get());
                 return ClusterState.builder(currentState).metaData(mdBuilder).build();
             }
 
@@ -142,7 +143,7 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
                 LicensesMetaData currentLicenses = metaData.custom(LicensesMetaData.TYPE);
                 final LicensesWrapper licensesWrapper = LicensesWrapper.wrap(currentLicenses);
                 licensesWrapper.removeFeatures(esLicenseManager, request.features());
-                mdBuilder.putCustom(LicensesMetaData.TYPE, licensesWrapper.createLicensesMetaData());
+                mdBuilder.putCustom(LicensesMetaData.TYPE, licensesWrapper.get());
                 return ClusterState.builder(currentState).metaData(mdBuilder).build();
             }
         });
@@ -182,24 +183,10 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
     @Override
     public List<ESLicense> getLicenses() {
         LicensesMetaData currentMetaData = clusterService.state().metaData().custom(LicensesMetaData.TYPE);
-        Set<ESLicense> trialLicenses = new HashSet<>();
         if (currentMetaData != null) {
-            Set<ESLicense> currentLicenses = esLicenseManager.fromSignatures(currentMetaData.getSignatures());
+            // don't use ESLicenses.reduceAndMap, as it will merge out expired licenses
-            TrialLicenses currentTrialLicenses = TrialLicenseUtils.fromEncodedTrialLicenses(currentMetaData.getEncodedTrialLicenses());
+            Set<ESLicense> licenses = Sets.union(esLicenseManager.fromSignatures(currentMetaData.getSignatures()),
-            for (TrialLicense trialLicense : currentTrialLicenses) {
+                    TrialLicenseUtils.fromEncodedTrialLicenses(currentMetaData.getEncodedTrialLicenses()));
-                trialLicenses.add(ESLicense.builder()
-                                .uid(trialLicense.uid())
-                                .issuedTo(trialLicense.issuedTo())
-                                .issueDate(trialLicense.issueDate())
-                                .type(ESLicense.Type.TRIAL)
-                                .subscriptionType(ESLicense.SubscriptionType.NONE)
-                                .feature(trialLicense.feature())
-                                .maxNodes(trialLicense.maxNodes())
-                                .expiryDate(trialLicense.expiryDate())
-                                .issuer("elasticsearch").buildInternal()
-                );
-            }
-            Set<ESLicense> licenses = Sets.union(currentLicenses, trialLicenses);
 
             // bucket license for feature with the latest expiry date
             Map<String, ESLicense> licenseMap = new HashMap<>();
@@ -244,11 +231,10 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
                 MetaData.Builder mdBuilder = MetaData.builder(currentState.metaData());
                 LicensesMetaData currentLicensesMetaData = metaData.custom(LicensesMetaData.TYPE);
                 final LicensesWrapper licensesWrapper = LicensesWrapper.wrap(currentLicensesMetaData);
-                // do not generate a trial license for a feature that already has a signed license
+                // do not generate a trial license for a feature that already has a signed/trial license
-                if (!hasLicenseForFeature(request.feature, currentLicensesMetaData)) {
+                licensesWrapper.addTrialLicenseIfNeeded(esLicenseManager,
-                    licensesWrapper.addTrialLicense(generateTrialLicense(request.feature, request.duration, request.maxNodes));
+                        generateTrialLicense(request.feature, request.duration, request.maxNodes));
-                }
+                mdBuilder.putCustom(LicensesMetaData.TYPE, licensesWrapper.get());
-                mdBuilder.putCustom(LicensesMetaData.TYPE, licensesWrapper.createLicensesMetaData());
                 return ClusterState.builder(currentState).metaData(mdBuilder).build();
             }
 
@@ -257,8 +243,8 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
                 logger.info("LicensesService: " + source, t);
             }
 
-            private TrialLicense generateTrialLicense(String feature, TimeValue duration, int maxNodes) {
+            private ESLicense generateTrialLicense(String feature, TimeValue duration, int maxNodes) {
-                return TrialLicensesBuilder.trialLicenseBuilder()
+                return TrialLicenseUtils.builder()
                         .issuedTo(clusterService.state().getClusterName().value())
                         .issueDate(System.currentTimeMillis())
                         .duration(duration)
@@ -267,8 +253,6 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
                         .build();
             }
         });
-
-
     }
 
     @Override
@@ -288,9 +272,9 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
     @Override
     protected void doClose() throws ElasticsearchException {
         logger.info("Closing LicensesService");
-        if (notificationScheduler != null) {
+        if (notificationScheduler.get() != null) {
-            notificationScheduler.cancel(true);
+            notificationScheduler.get().cancel(true);
-            notificationScheduler = null;
+            notificationScheduler.set(null);
         }
         clusterService.remove(this);
 
@@ -302,12 +286,15 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
             // clear all handlers
             registeredListeners.clear();
         }
+
+        lastObservedState.set(null);
     }
 
     @Override
     public void clusterChanged(ClusterChangedEvent event) {
         if (!event.state().blocks().hasGlobalBlock(GatewayService.STATE_NOT_RECOVERED_BLOCK)) {
 
+            // Check pending feature registrations and try to complete registrations
             if (!pendingRegistrations.isEmpty()) {
                 ListenerHolder pendingRegistrationLister;
                 while ((pendingRegistrationLister = pendingRegistrations.poll()) != null) {
@@ -321,23 +308,43 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
             }
 
             // notify all interested plugins
-            LicensesMetaData currentLicensesMetaData = event.state().getMetaData().custom(LicensesMetaData.TYPE);
+            // notifyFeaturesIfNeeded will short-circuit with -1 if the currentLicensesMetaData has been notified on earlier
-
-            // checkIfUpdatedMetaData should be called to see if the license metaData has changed,
-            // but upon registration, the oldest cluster state might have a license
-
             // Change to debug
-            logger.info("calling notifyFeatures from clusterChanged");
+            logger.info("calling notifyFeaturesAndScheduleNotificationIfNeeded from clusterChanged");
-            long nextScheduleFrequency = notifyFeatures(currentLicensesMetaData);
+            LicensesMetaData currentLicensesMetaData = event.state().getMetaData().custom(LicensesMetaData.TYPE);
-            if (notificationScheduler == null) {
+            notifyFeaturesAndScheduleNotificationIfNeeded(currentLicensesMetaData);
-                notificationScheduler = threadPool.schedule(TimeValue.timeValueMillis(nextScheduleFrequency), executorName(),
-                        new SubmitReschedulingLicensingClientNotificationJob());
-            }
         } else {
             logger.info("clusterChanged: no action [has STATE_NOT_RECOVERED_BLOCK]");
         }
     }
 
+    private void notifyFeaturesAndScheduleNotificationIfNeeded(LicensesMetaData currentLicensesMetaData) {
+        final LicensesMetaData lastNotifiedLicensesMetaData = lastObservedState.get();
+        if (lastNotifiedLicensesMetaData != null && lastNotifiedLicensesMetaData.equals(currentLicensesMetaData)) {
+            logger.info("currentLicensesMetaData has been already notified on");
+            return;
+        }
+        notifyFeaturesAndScheduleNotification(currentLicensesMetaData);
+    }
+
+    private long notifyFeaturesAndScheduleNotification(LicensesMetaData currentLicensesMetaData) {
+        long nextScheduleFrequency = notifyFeatures(currentLicensesMetaData);
+        logger.info("Condition to register new notification schedule: null notification: " + (notificationScheduler.get() == null) + " , nextScheduleFreq: " + (nextScheduleFrequency != -1));
+        if (notificationScheduler.get() == null && nextScheduleFrequency != -1l) {
+            logger.info("enabling licensing client notifications");
+            notificationScheduler.set(threadPool.schedule(TimeValue.timeValueMillis(nextScheduleFrequency), executorName(),
+                    new SubmitReschedulingLicensingClientNotificationJob()));
+        } else {
+            if (notificationScheduler.get() != null) {
+                logger.info("disable license client notification");
+                notificationScheduler.get().cancel(true);
+                // set it to null so that new notifications can be scheduled on licensesMetaData change (cluster state change) if needed
+                notificationScheduler.set(null);
+            }
+        }
+        return nextScheduleFrequency;
+    }
+
     private void logLicenseMetaDataStats(String prefix, LicensesMetaData licensesMetaData) {
         if (licensesMetaData != null) {
             logger.info(prefix + " LicensesMetaData: signedLicenses: " + licensesMetaData.getSignatures().size() + " trialLicenses: " + licensesMetaData.getEncodedTrialLicenses().size());
@@ -346,23 +353,6 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
         }
     }
 
-    private LicensesMetaData checkIfUpdatedMetaData(ClusterChangedEvent event) {
-        LicensesMetaData oldMetaData = event.previousState().getMetaData().custom(LicensesMetaData.TYPE);
-        LicensesMetaData newMetaData = event.state().getMetaData().custom(LicensesMetaData.TYPE);
-
-        logLicenseMetaDataStats("old", oldMetaData);
-        logLicenseMetaDataStats("new", newMetaData);
-
-        if ((oldMetaData == null && newMetaData == null) || (oldMetaData != null && oldMetaData.equals(newMetaData))) {
-            logger.info("no change in LicensesMetaData");
-            return null;
-        } else {
-            logger.info("detected change in LicensesMetaData");
-            return newMetaData;
-        }
-    }
-
-
     @Override
     public void register(String feature, TrialLicenseOptions trialLicenseOptions, Listener listener) {
         final ListenerHolder listenerHolder = new ListenerHolder(feature, trialLicenseOptions, listener);
@@ -385,12 +375,12 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
 
         LicensesMetaData currentMetaData = clusterService.state().metaData().custom(LicensesMetaData.TYPE);
         if (!hasLicenseForFeature(listenerHolder.feature, currentMetaData)) {
-            // does not have actual license so generate a trial license
+            // does not have any license so generate a trial license
             TrialLicenseOptions options = listenerHolder.trialLicenseOptions;
             if (options != null) {
                 // Trial license option is provided
                 RegisterTrialLicenseRequest request = new RegisterTrialLicenseRequest(listenerHolder.feature,
-                        new TimeValue(options.durationInDays, TimeUnit.DAYS), options.maxNodes);
+                        options.duration, options.maxNodes);
                 if (clusterService.state().nodes().localNodeMaster()) {
                     logger.info("Executing trial license request");
                     registerTrialLicense(request);
@@ -411,20 +401,27 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
                 // notify feature as clusterChangedEvent may not happen
                 // as no trial or signed license has been found for feature
                 // Change to debug
-                logger.info("Calling notifyFeatures [no trial license spec provided]");
+                logger.info("Calling notifyFeaturesAndScheduleNotification [no trial license spec provided]");
-                notifyFeatures(currentMetaData);
+                notifyFeaturesAndScheduleNotification(currentMetaData);
             }
         } else {
             // signed license already found for the new registered
             // feature, notify feature on registration
-            logger.info("Calling notifyFeatures [signed license available]");
+            logger.info("Calling notifyFeaturesAndScheduleNotification [signed/trial license available]");
-            notifyFeatures(currentMetaData);
+            notifyFeaturesAndScheduleNotification(currentMetaData);
         }
         return true;
     }
 
     private boolean hasLicenseForFeature(String feature, LicensesMetaData currentLicensesMetaData) {
-        return esLicenseManager.hasLicenseForFeature(feature, getEffectiveLicenses(currentLicensesMetaData));
+        final Map<String, ESLicense> effectiveLicenses = getEffectiveLicenses(currentLicensesMetaData);
+        ESLicense featureLicense;
+        if ((featureLicense = effectiveLicenses.get(feature)) != null) {
+            if (featureLicense.expiryDate() > System.currentTimeMillis()) {
+                return true;
+            }
+        }
+        return false;
     }
 
 
@@ -432,13 +429,12 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
         return ThreadPool.Names.GENERIC;
     }
 
-    public Map<String, ESLicense> getEffectiveLicenses(LicensesMetaData metaData) {
+    private Map<String, ESLicense> getEffectiveLicenses(LicensesMetaData metaData) {
         Map<String, ESLicense> map = new HashMap<>();
         if (metaData != null) {
             Set<ESLicense> esLicenses = new HashSet<>();
-            for (String signature : metaData.getSignatures()) {
+            esLicenses.addAll(esLicenseManager.fromSignatures(metaData.getSignatures()));
-                esLicenses.add(esLicenseManager.fromSignature(signature));
+            esLicenses.addAll(TrialLicenseUtils.fromEncodedTrialLicenses(metaData.getEncodedTrialLicenses()));
-            }
             return reduceAndMap(esLicenses);
         }
         return ImmutableMap.copyOf(map);
@@ -452,11 +448,9 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
                 logger.trace("Submitting new rescheduling licensing client notification job");
             }
             try {
-                threadPool.executor(executorName()).execute(new LicensingClientNotificationJob(true));
+                threadPool.executor(executorName()).execute(new LicensingClientNotificationJob());
             } catch (EsRejectedExecutionException ex) {
-                if (logger.isDebugEnabled()) {
+                logger.info("Couldn't re-schedule licensing client notification job", ex);
-                    logger.debug("Couldn't re-schedule licensing client notification job", ex);
-                }
             }
         }
     }
@@ -468,10 +462,7 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
 
     public class LicensingClientNotificationJob implements Runnable {
 
-        private final boolean reschedule;
+        public LicensingClientNotificationJob() {
-
-        public LicensingClientNotificationJob(boolean reschedule) {
-            this.reschedule = reschedule;
         }
 
         @Override
@@ -480,24 +471,22 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
                 logger.trace("Performing LicensingClientNotificationJob");
             }
 
-            if (clusterService.state().nodes().localNodeMaster()) {
+            LicensesMetaData currentLicensesMetaData = clusterService.state().metaData().custom(LicensesMetaData.TYPE);
-                LicensesMetaData currentLicensesMetaData = clusterService.state().metaData().custom(LicensesMetaData.TYPE);
 
-                // Change to debug
+            // Change to debug
-                logger.info("calling notifyFeatures from LicensingClientNotificationJob");
+            logger.info("calling notifyFeaturesIfNeeded from LicensingClientNotificationJob");
-                long nextScheduleFrequency = Math.max(TimeValue.timeValueMinutes(5).getMillis(), notifyFeatures(currentLicensesMetaData));
-                TimeValue updateFrequency = TimeValue.timeValueMillis(nextScheduleFrequency);
 
-                if (this.reschedule) {
+            long nextScheduleFrequency;
-                    if (logger.isTraceEnabled()) {
+            if ((nextScheduleFrequency = notifyFeaturesAndScheduleNotification(currentLicensesMetaData)) == -1l) {
-                        logger.trace("Scheduling next run for licensing client notification job in: {}", updateFrequency.toString());
+                return;
-                    }
+            }
-                    try {
+
-                        threadPool.schedule(updateFrequency, executorName(), new SubmitReschedulingLicensingClientNotificationJob());
+            TimeValue updateFrequency = TimeValue.timeValueMillis(nextScheduleFrequency);
-                    } catch (EsRejectedExecutionException ex) {
+            logger.trace("Scheduling next run for licensing client notification job in: {}", updateFrequency.toString());
-                        logger.debug("Reschedule licensing client notification job was rejected", ex);
+            try {
-                    }
+                threadPool.schedule(updateFrequency, executorName(), new SubmitReschedulingLicensingClientNotificationJob());
-                }
+            } catch (EsRejectedExecutionException ex) {
+                logger.info("Reschedule licensing client notification job was rejected", ex);
             }
         }
     }
@@ -505,7 +494,7 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
     private long notifyFeatures(LicensesMetaData currentLicensesMetaData) {
         LicensesWrapper licensesWrapper = LicensesWrapper.wrap(currentLicensesMetaData);
         long nextScheduleFrequency = -1l;
-        long offset = TimeValue.timeValueMinutes(1).getMillis();
+        long offset = TimeValue.timeValueMillis(100).getMillis();
         StringBuilder sb = new StringBuilder("Registered listeners: [ ");
         for (ListenerHolder listenerHolder : registeredListeners) {
 
@@ -517,20 +506,13 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
             long expiryDate = -1l;
             if (hasLicenseForFeature(listenerHolder.feature, currentLicensesMetaData)) {
                 final Map<String, ESLicense> effectiveLicenses = getEffectiveLicenses(currentLicensesMetaData);
-                expiryDate = effectiveLicenses.get(listenerHolder.feature).expiryDate();
+                final ESLicense license = effectiveLicenses.get(listenerHolder.feature);
+                expiryDate = license.expiryDate();
 
-                sb.append("signed license expiry: ");
+                sb.append((license.signature() != null) ? "signed" : "trial");
+                sb.append(" license expiry: ");
                 sb.append(expiryDate);
                 sb.append(", ");
-            } else {
-                final TrialLicense trialLicense = licensesWrapper.trialLicenses().getTrialLicense(listenerHolder.feature);
-                if (trialLicense != null) {
-                    expiryDate = trialLicense.expiryDate();
-
-                    sb.append("trial license expiry: ");
-                    sb.append(expiryDate);
-                    sb.append(", ");
-                }
             }
             long expiryDuration = expiryDate - System.currentTimeMillis();
 
@@ -555,25 +537,23 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
             } else {
                 // Change to debug
                 sb.append("calling disableFeatureIfNeeded");
-
                 listenerHolder.disableFeatureIfNeeded();
             }
-
             sb.append(" )");
         }
         sb.append("]");
         logger.info(sb.toString());
 
+        lastObservedState.set(licensesWrapper.get());
+
         if (nextScheduleFrequency == -1l) {
-            nextScheduleFrequency = TimeValue.timeValueMinutes(5).getMillis();
+            logger.info("turn off notifications");
-            logger.info("next notification time set to default of 5 minutes");
         } else {
             logger.info("next notification time: " + TimeValue.timeValueMillis(nextScheduleFrequency).toString());
         }
 
-
-
         return nextScheduleFrequency;
+
     }
 
     public static class PutLicenseRequestHolder {
@@ -596,19 +576,17 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
         }
     }
 
-
-
     public static class TrialLicenseOptions {
-        final int durationInDays;
+        final TimeValue duration;
         final int maxNodes;
 
-        public TrialLicenseOptions(int durationInDays, int maxNodes) {
+        public TrialLicenseOptions(TimeValue duration, int maxNodes) {
-            this.durationInDays = durationInDays;
+            this.duration = duration;
             this.maxNodes = maxNodes;
         }
     }
 
-    private static class ListenerHolder {
+    private class ListenerHolder {
         final String feature;
         final TrialLicenseOptions trialLicenseOptions;
         final Listener listener;
@@ -622,13 +600,17 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
         }
 
         private void enableFeatureIfNeeded() {
+            logger.info("enabled flag: " + enabled.get());
             if (enabled.compareAndSet(false, true)) {
+                logger.info("calling onEnabled on listener");
                 listener.onEnabled();
             }
         }
 
         private void disableFeatureIfNeeded() {
+            logger.info("enabled flag: " + enabled.get());
             if (enabled.compareAndSet(true, false)) {
+                logger.info("calling onDisabled on listener");
                 listener.onDisabled();
             }
         }
@@ -654,7 +636,7 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
             return licenseManager.fromSignatures(signatures);
         }
 
-        public TrialLicenses trialLicenses() {
+        public Set<ESLicense> trialLicenses() {
             return TrialLicenseUtils.fromEncodedTrialLicenses(encodedTrialLicenses);
         }
 
@@ -665,17 +647,17 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
          *
          * @param trialLicense to add
          */
-        public void addTrialLicense(TrialLicense trialLicense) {
+        public void addTrialLicenseIfNeeded(ESLicenseManager licenseManager, ESLicense generatedTrialLicense) {
             boolean featureTrialLicenseExists = false;
-            for (TrialLicense currentTrialLicense : trialLicenses()) {
+            for (ESLicense license : Sets.union(signedLicenses(licenseManager),trialLicenses())) {
-                if (currentTrialLicense.feature().equals(trialLicense.feature())) {
+                if (license.feature().equals(generatedTrialLicense.feature())) {
                     featureTrialLicenseExists = true;
                     break;
                 }
             }
             if (!featureTrialLicenseExists) {
                 this.encodedTrialLicenses = ImmutableSet.copyOf(Sets.union(encodedTrialLicenses,
-                        Collections.singleton(TrialLicenseUtils.toEncodedTrialLicense(trialLicense))));
+                        Collections.singleton(TrialLicenseUtils.toEncodedTrialLicense(generatedTrialLicense))));
             }
         }
 
@@ -698,12 +680,11 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
             this.signatures = licenseManager.toSignatures(reducedLicenses);
         }
 
-        public LicensesMetaData createLicensesMetaData() {
+        public LicensesMetaData get() {
             return new LicensesMetaData(signatures, encodedTrialLicenses);
         }
     }
 
-
     private static class RegisterTrialLicenseRequest extends TransportRequest {
         private int maxNodes;
         private String feature;
@@ -736,7 +717,6 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
         }
     }
 
-
     private class RegisterTrialLicenseRequestHandler extends BaseTransportRequestHandler<RegisterTrialLicenseRequest> {
         @Override
         public RegisterTrialLicenseRequest newInstance() {
@@ -757,11 +737,10 @@ public class LicensesService extends AbstractLifecycleComponent<LicensesService>
 
     //Should not be exposed; used by testing only
     public void clear() {
-        if (notificationScheduler != null) {
+        if (notificationScheduler.get() != null) {
-            notificationScheduler.cancel(true);
+            notificationScheduler.get().cancel(true);
-            notificationScheduler = null;
+            notificationScheduler.set(null);
         }
         registeredListeners.clear();
-
     }
 }

src/main/java/org/elasticsearch/license/plugin/core/TrialLicenseUtils.java

@@ -0,0 +1,167 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License;
+ * you may not use this file except in compliance with the Elastic License.
+ */
+package org.elasticsearch.license.plugin.core;
+
+import org.apache.commons.codec.binary.Base64;
+import org.elasticsearch.common.collect.ImmutableSet;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.license.core.ESLicense;
+import org.elasticsearch.license.core.LicensesCharset;
+
+import java.nio.ByteBuffer;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.UUID;
+
+import static org.elasticsearch.license.core.ESLicense.SubscriptionType;
+import static org.elasticsearch.license.core.ESLicense.Type;
+
+public class TrialLicenseUtils {
+
+    public static TrialLicenseBuilder builder() {
+        return new TrialLicenseBuilder();
+    }
+
+    public static class TrialLicenseBuilder {
+        private static final String DEFAULT_ISSUER = "elasticsearch";
+        private static final Type DEFAULT_TYPE = Type.TRIAL;
+        private static final SubscriptionType DEFAULT_SUBSCRIPTION_TYPE = SubscriptionType.NONE;
+
+        private String feature;
+        private long expiryDate = -1;
+        private long issueDate = -1;
+        private TimeValue duration;
+        private int maxNodes = -1;
+        private String uid = null;
+        private String issuedTo;
+
+        public TrialLicenseBuilder() {
+        }
+
+        public TrialLicenseBuilder uid(String uid) {
+            this.uid = uid;
+            return this;
+        }
+
+        public TrialLicenseBuilder issuedTo(String issuedTo) {
+            this.issuedTo = issuedTo;
+            return this;
+        }
+
+        public TrialLicenseBuilder maxNodes(int maxNodes) {
+            this.maxNodes = maxNodes;
+            return this;
+        }
+
+        public TrialLicenseBuilder feature(String featureType) {
+            this.feature = featureType;
+            return this;
+        }
+
+        public TrialLicenseBuilder issueDate(long issueDate) {
+            this.issueDate = issueDate;
+            return this;
+        }
+
+        public TrialLicenseBuilder duration(TimeValue duration) {
+            this.duration = duration;
+            return this;
+        }
+
+        public TrialLicenseBuilder expiryDate(long expiryDate) {
+            this.expiryDate = expiryDate;
+            return this;
+        }
+
+        public ESLicense build() {
+            if (expiryDate == -1) {
+                expiryDate = issueDate + duration.millis();
+            }
+            if (uid == null) {
+                uid = UUID.randomUUID().toString();
+            }
+            return ESLicense.builder()
+                                .type(DEFAULT_TYPE)
+                                .subscriptionType(DEFAULT_SUBSCRIPTION_TYPE)
+                                .issuer(DEFAULT_ISSUER)
+                                .uid(uid)
+                                .issuedTo(issuedTo)
+                                .issueDate(issueDate)
+                                .feature(feature)
+                                .maxNodes(maxNodes)
+                                .expiryDate(expiryDate)
+                                .build();
+        }
+
+    }
+
+    public static Set<ESLicense> fromEncodedTrialLicenses(Set<String> encodedTrialLicenses) {
+        Set<ESLicense> licenses = new HashSet<>(encodedTrialLicenses.size());
+        for (String encodedTrialLicense : encodedTrialLicenses) {
+            licenses.add(fromEncodedTrialLicense(encodedTrialLicense));
+        }
+        return ImmutableSet.copyOf(licenses);
+    }
+
+    public static ESLicense fromEncodedTrialLicense(String encodedTrialLicense) {
+        byte[] encodedBytes = Base64.decodeBase64(encodedTrialLicense);
+        ByteBuffer byteBuffer = ByteBuffer.wrap(encodedBytes);
+
+        int uidLen = byteBuffer.getInt();
+        byte[] uidBytes = new byte[uidLen];
+        byteBuffer.get(uidBytes);
+        String uid = new String(uidBytes, LicensesCharset.UTF_8);
+
+        int issuedToLen = byteBuffer.getInt();
+        byte[] issuedToBytes = new byte[issuedToLen];
+        byteBuffer.get(issuedToBytes);
+        String issuedTo = new String(issuedToBytes, LicensesCharset.UTF_8);
+
+        int featureLen = byteBuffer.getInt();
+        byte[] featureBytes = new byte[featureLen];
+        byteBuffer.get(featureBytes);
+        String feature = new String(featureBytes, LicensesCharset.UTF_8);
+
+        int maxNodes = byteBuffer.getInt();
+        long issueDate = byteBuffer.getLong();
+        long expiryDate = byteBuffer.getLong();
+
+        return builder()
+                .uid(uid)
+                .issuedTo(issuedTo)
+                .feature(feature)
+                .maxNodes(maxNodes)
+                .issueDate(issueDate)
+                .expiryDate(expiryDate)
+                .build();
+    }
+
+    public static String toEncodedTrialLicense(ESLicense trialLicense) {
+        byte[] uidBytes = trialLicense.uid().getBytes(LicensesCharset.UTF_8);
+        byte[] featureBytes = trialLicense.feature().getBytes(LicensesCharset.UTF_8);
+        byte[] issuedToBytes = trialLicense.issuedTo().getBytes(LicensesCharset.UTF_8);
+
+        // uid len + uid bytes + issuedTo len + issuedTo bytes + feature bytes length + feature bytes + maxNodes + issueDate + expiryDate
+        int len = 4 + uidBytes.length + 4 + issuedToBytes.length + 4 + featureBytes.length + 4 + 8 + 8;
+        final byte[] encodedLicense = new byte[len];
+        ByteBuffer byteBuffer = ByteBuffer.wrap(encodedLicense);
+
+        byteBuffer.putInt(uidBytes.length);
+        byteBuffer.put(uidBytes);
+
+        byteBuffer.putInt(issuedToBytes.length);
+        byteBuffer.put(issuedToBytes);
+
+        byteBuffer.putInt(featureBytes.length);
+        byteBuffer.put(featureBytes);
+
+        byteBuffer.putInt(trialLicense.maxNodes());
+        byteBuffer.putLong(trialLicense.issueDate());
+        byteBuffer.putLong(trialLicense.expiryDate());
+
+        return Base64.encodeBase64String(encodedLicense);
+    }
+}

src/test/java/org/elasticsearch/license/TestUtils.java

@@ -8,8 +8,8 @@ package org.elasticsearch.license;
 import org.apache.commons.io.FileUtils;
 import org.elasticsearch.license.core.DateUtils;
 import org.elasticsearch.license.core.ESLicense;
+import org.elasticsearch.license.core.ESLicenses;
 import org.elasticsearch.license.licensor.tools.LicenseGeneratorTool;
-import org.elasticsearch.license.manager.Utils;
 
 import java.io.File;
 import java.io.FileOutputStream;
@@ -76,7 +76,7 @@ public class TestUtils {
     }
 
     public static void verifyESLicenses(Set<ESLicense> esLicenses, Map<String, FeatureAttributes> featureAttributesMap) throws ParseException {
-        verifyESLicenses(Utils.reduceAndMap(esLicenses), featureAttributesMap);
+        verifyESLicenses(ESLicenses.reduceAndMap(esLicenses), featureAttributesMap);
 
     }
 
@@ -106,9 +106,9 @@ public class TestUtils {
 
     public static void isSame(Set<ESLicense> firstLicenses, Set<ESLicense> secondLicenses) {
 
-        // we do the build to make sure we weed out any expired licenses
+        // we do the verifyAndBuild to make sure we weed out any expired licenses
-        final Map<String, ESLicense> licenses1 = Utils.reduceAndMap(firstLicenses);
+        final Map<String, ESLicense> licenses1 = ESLicenses.reduceAndMap(firstLicenses);
-        final Map<String, ESLicense> licenses2 = Utils.reduceAndMap(secondLicenses);
+        final Map<String, ESLicense> licenses2 = ESLicenses.reduceAndMap(secondLicenses);
 
         // check if the effective licenses have the same feature set
         assertTrue("Both licenses should have the same number of features", licenses1.size() == licenses2.size());

src/test/java/org/elasticsearch/license/manager/LicenseVerificationTests.java

@@ -71,7 +71,7 @@ public class LicenseVerificationTests extends AbstractLicensingTestBase {
 
         esLicenseManager.verifyLicenses(esLicenseProvider.getEffectiveLicenses());
 
-        verifyLicenseManager(esLicenseManager, esLicenseProvider, map);
+        verifyLicenses(esLicenseProvider, map);
 
     }
 
@@ -95,7 +95,7 @@ public class LicenseVerificationTests extends AbstractLicensingTestBase {
 
         esLicenseManager.verifyLicenses(esLicenseProvider.getEffectiveLicenses());
 
-        verifyLicenseManager(esLicenseManager, esLicenseProvider, map);
+        verifyLicenses(esLicenseProvider, map);
 
     }
 
@@ -120,9 +120,8 @@ public class LicenseVerificationTests extends AbstractLicensingTestBase {
 
         // All validation for shield license should be normal as expected
 
-        verifyLicenseManager(esLicenseManager, esLicenseProvider, Collections.singletonMap(TestUtils.SHIELD, shildFeatureAttributes));
+        verifyLicenses(esLicenseProvider, Collections.singletonMap(TestUtils.SHIELD, shildFeatureAttributes));
 
-        assertFalse("license for marvel should not be valid due to expired expiry date", esLicenseManager.hasLicenseForFeature(TestUtils.MARVEL, esLicenseProvider.getEffectiveLicenses()));
     }
 
     @Test
@@ -137,14 +136,14 @@ public class LicenseVerificationTests extends AbstractLicensingTestBase {
 
         Set<ESLicense> esLicensesOutput = new HashSet<>(ESLicenses.fromSource(generateSignedLicenses(map)));
 
-        ESLicense esLicense = Utils.reduceAndMap(esLicensesOutput).get(TestUtils.SHIELD);
+        ESLicense esLicense = ESLicenses.reduceAndMap(esLicensesOutput).get(TestUtils.SHIELD);
 
         final ESLicense tamperedLicense = ESLicense.builder()
                 .fromLicense(esLicense)
                 .expiryDate(esLicense.expiryDate() + 10 * 24 * 60 * 60 * 1000l)
                 .feature(TestUtils.SHIELD)
                 .issuer("elasticsqearch")
-                .build();
+                .verifyAndBuild();
 
         try {
             esLicenseProvider.setLicenses(Collections.singleton(tamperedLicense));
@@ -155,7 +154,7 @@ public class LicenseVerificationTests extends AbstractLicensingTestBase {
         }
     }
 
-    public static void verifyLicenseManager(ESLicenseManager esLicenseManager, FileBasedESLicenseProvider licenseProvider, Map<String, TestUtils.FeatureAttributes> featureAttributeMap) throws ParseException {
+    public static void verifyLicenses(FileBasedESLicenseProvider licenseProvider, Map<String, TestUtils.FeatureAttributes> featureAttributeMap) throws ParseException {
 
         for (Map.Entry<String, TestUtils.FeatureAttributes> entry : featureAttributeMap.entrySet()) {
             TestUtils.FeatureAttributes featureAttributes = entry.getValue();
@@ -169,7 +168,6 @@ public class LicenseVerificationTests extends AbstractLicensingTestBase {
             assertTrue("License should have subscription type of " + featureAttributes.subscriptionType, license.subscriptionType() == ESLicense.SubscriptionType.fromString(featureAttributes.subscriptionType));
 
 
-            assertTrue("License should be valid for " + featureType, esLicenseManager.hasLicenseForFeature(featureType, licenseProvider.getEffectiveLicenses()));
             assertTrue("License should be valid for maxNodes = " + (featureAttributes.maxNodes), license.maxNodes() == featureAttributes.maxNodes);
         }
     }

src/test/java/org/elasticsearch/license/plugin/LicenseTransportTests.java

@@ -13,7 +13,6 @@ import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.license.TestUtils;
 import org.elasticsearch.license.core.ESLicense;
 import org.elasticsearch.license.core.ESLicenses;
-import org.elasticsearch.license.manager.Utils;
 import org.elasticsearch.license.plugin.action.delete.DeleteLicenseRequestBuilder;
 import org.elasticsearch.license.plugin.action.delete.DeleteLicenseResponse;
 import org.elasticsearch.license.plugin.action.get.GetLicenseRequestBuilder;
@@ -122,7 +121,7 @@ public class LicenseTransportTests extends ElasticsearchIntegrationTest {
         assertTrue(deleteLicenseResponse.isAcknowledged());
 
         //getLicenseResponse = new GetLicenseRequestBuilder(client().admin().cluster()).execute().get();
-        //TestUtils.isSame(getLicenseResponse.licenses(), LicenseBuilders.licensesBuilder().build());
+        //TestUtils.isSame(getLicenseResponse.licenses(), LicenseBuilders.licensesBuilder().verifyAndBuild());
     }
 
     @Test
@@ -136,14 +135,14 @@ public class LicenseTransportTests extends ElasticsearchIntegrationTest {
 
         Set<ESLicense> esLicenses = new HashSet<>(ESLicenses.fromSource(licenseOutput));
 
-        ESLicense esLicense = Utils.reduceAndMap(esLicenses).get(TestUtils.SHIELD);
+        ESLicense esLicense = ESLicenses.reduceAndMap(esLicenses).get(TestUtils.SHIELD);
 
         final ESLicense tamperedLicense = ESLicense.builder()
                 .fromLicense(esLicense)
                 .expiryDate(esLicense.expiryDate() + 10 * 24 * 60 * 60 * 1000l)
                 .feature(TestUtils.SHIELD)
                 .issuer("elasticsqearch")
-                .build();
+                .verifyAndBuild();
 
         PutLicenseRequestBuilder builder = new PutLicenseRequestBuilder(client().admin().cluster());
         builder.setLicense(Collections.singletonList(tamperedLicense));

src/test/java/org/elasticsearch/license/plugin/LicensesPluginIntegrationTests.java

@@ -5,24 +5,36 @@
  */
 package org.elasticsearch.license.plugin;
 
+import org.elasticsearch.common.base.Predicate;
 import org.elasticsearch.common.settings.ImmutableSettings;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.license.plugin.core.LicensesManagerService;
 import org.elasticsearch.test.ElasticsearchIntegrationTest;
 import org.elasticsearch.test.InternalTestCluster;
+import org.junit.Before;
 import org.junit.Test;
 
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
 import static org.elasticsearch.test.ElasticsearchIntegrationTest.ClusterScope;
 import static org.elasticsearch.test.ElasticsearchIntegrationTest.Scope.SUITE;
+import static org.elasticsearch.test.ElasticsearchIntegrationTest.Scope.TEST;
+import static org.hamcrest.CoreMatchers.equalTo;
 
-@ClusterScope(scope = SUITE, numDataNodes = 10)
+@ClusterScope(scope = TEST, numDataNodes = 10, numClientNodes = 0)
 public class LicensesPluginIntegrationTests extends ElasticsearchIntegrationTest {
 
+    private final int trialLicenseDurationInSeconds = 5;
 
     @Override
     protected Settings nodeSettings(int nodeOrdinal) {
         return ImmutableSettings.settingsBuilder()
                 .put("plugins.load_classpath_plugins", false)
+                .put("test_consumer_plugin.trial_license_duration_in_seconds", trialLicenseDurationInSeconds)
                 .put("plugin.types", LicensePlugin.class.getName() + "," + TestConsumerPlugin.class.getName())
                 .build();
     }
@@ -34,23 +46,62 @@ public class LicensesPluginIntegrationTests extends ElasticsearchIntegrationTest
     }
 
     @Test
-    public void testLicenseRegistration() throws Exception {
+    public void test() throws InterruptedException {
-        LicensesManagerService managerService = licensesManagerService();
+        // managerService should report feature to be enabled on all data nodes
-        assertTrue(managerService.enabledFeatures().contains(TestPluginService.FEATURE_NAME));
+        assertThat(awaitBusy(new Predicate<Object>() {
+            @Override
+            public boolean apply(Object o) {
+                for (LicensesManagerService managerService : licensesManagerServices()) {
+                    if (!managerService.enabledFeatures().contains(TestPluginService.FEATURE_NAME)) {
+                        return false;
+                    }
+                }
+                return true;
+            }
+        }, 2, TimeUnit.SECONDS), equalTo(true));
+
+
+        // consumer plugin service should return enabled on all data nodes
+        assertThat(awaitBusy(new Predicate<Object>() {
+            @Override
+            public boolean apply(Object o) {
+                for (TestPluginService pluginService : consumerPluginServices()) {
+                    if (!pluginService.enabled()) {
+                        return false;
+                    }
+                }
+                return true;
+            }
+        }, 2, TimeUnit.SECONDS), equalTo(true));
+
+
+        // consumer plugin should notify onDisabled on all data nodes (expired trial license)
+        assertThat(awaitBusy(new Predicate<Object>() {
+            @Override
+            public boolean apply(Object o) {
+                for (TestPluginService pluginService : consumerPluginServices()) {
+                    if(pluginService.enabled()) {
+                        return false;
+                    }
+
+                }
+                return true;
+            }
+        }, trialLicenseDurationInSeconds * 2, TimeUnit.SECONDS), equalTo(true));
+
     }
 
-    @Test
+    private Iterable<TestPluginService> consumerPluginServices() {
-    public void testFeatureActivation() throws Exception {
-        TestPluginService pluginService = consumerPluginService();
-        assertTrue(pluginService.enabled());
-    }
-
-    private TestPluginService consumerPluginService() {
         final InternalTestCluster clients = internalCluster();
-        return clients.getInstance(TestPluginService.class, clients.getMasterName());
+        return clients.getDataNodeInstances(TestPluginService.class);
     }
 
-    private LicensesManagerService licensesManagerService() {
+    private Iterable<LicensesManagerService> licensesManagerServices() {
+        final InternalTestCluster clients = internalCluster();
+        return clients.getDataNodeInstances(LicensesManagerService.class);
+    }
+
+    private LicensesManagerService masterLicenseManagerService() {
         final InternalTestCluster clients = internalCluster();
         return clients.getInstance(LicensesManagerService.class, clients.getMasterName());
     }

src/test/java/org/elasticsearch/license/plugin/LicensesServiceNodeTests.java

@@ -10,6 +10,7 @@ import org.elasticsearch.common.settings.ImmutableSettings;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.node.internal.InternalNode;
 import org.elasticsearch.test.ElasticsearchIntegrationTest;
+import org.elasticsearch.test.junit.annotations.TestLogging;
 import org.junit.Test;
 
 import java.util.concurrent.TimeUnit;
@@ -19,13 +20,18 @@ import static org.hamcrest.Matchers.equalTo;
 
 /**
  */
-@ElasticsearchIntegrationTest.ClusterScope(scope = TEST, numDataNodes = 10)
+@ElasticsearchIntegrationTest.ClusterScope(scope = TEST, numDataNodes = 10, numClientNodes = 0)
 public class LicensesServiceNodeTests extends ElasticsearchIntegrationTest {
 
     @Override
     protected Settings nodeSettings(int nodeOrdinal) {
+        return nodeSettings();
+    }
+
+    private Settings nodeSettings() {
         return ImmutableSettings.settingsBuilder()
                 .put("plugins.load_classpath_plugins", false)
+                .put("test_consumer_plugin.trial_license_duration_in_seconds", 10)
                 .putArray("plugin.types", LicensePlugin.class.getName(), TestConsumerPlugin.class.getName())
                 .put(InternalNode.HTTP_ENABLED, true)
                 .build();
@@ -34,14 +40,14 @@ public class LicensesServiceNodeTests extends ElasticsearchIntegrationTest {
     @Override
     protected Settings transportClientSettings() {
         // Plugin should be loaded on the transport client as well
-        return nodeSettings(0);
+        return nodeSettings();
     }
 
 
     @Test
+    @TestLogging("_root:DEBUG")
     public void testPluginStatus() throws Exception {
         final Iterable<TestPluginService> testPluginServices = internalCluster().getDataNodeInstances(TestPluginService.class);
-
         assertThat(awaitBusy(new Predicate<Object>() {
             @Override
             public boolean apply(Object o) {
@@ -52,6 +58,8 @@ public class LicensesServiceNodeTests extends ElasticsearchIntegrationTest {
                 }
                 return true;
             }
-        }, 10, TimeUnit.SECONDS), equalTo(true));
+        }, 1, TimeUnit.MINUTES), equalTo(true));
+
     }
+
 }

src/test/java/org/elasticsearch/license/plugin/LicensesServiceTests.java

@@ -14,13 +14,14 @@ import org.elasticsearch.cluster.metadata.MetaData;
 import org.elasticsearch.cluster.node.DiscoveryNode;
 import org.elasticsearch.cluster.node.DiscoveryNodes;
 import org.elasticsearch.common.Nullable;
+import org.elasticsearch.common.base.Predicate;
 import org.elasticsearch.common.settings.ImmutableSettings;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.license.TestUtils;
 import org.elasticsearch.license.core.ESLicense;
 import org.elasticsearch.license.core.ESLicenses;
 import org.elasticsearch.license.manager.ESLicenseManager;
-import org.elasticsearch.license.manager.Utils;
 import org.elasticsearch.license.plugin.action.put.PutLicenseRequest;
 import org.elasticsearch.license.plugin.core.*;
 import org.elasticsearch.test.ElasticsearchIntegrationTest;
@@ -32,10 +33,12 @@ import java.net.URISyntaxException;
 import java.nio.file.Paths;
 import java.util.*;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.elasticsearch.test.ElasticsearchIntegrationTest.ClusterScope;
 import static org.elasticsearch.test.ElasticsearchIntegrationTest.Scope.TEST;
+import static org.hamcrest.Matchers.equalTo;
 
 @ClusterScope(scope = TEST, numDataNodes = 10)
 public class LicensesServiceTests extends ElasticsearchIntegrationTest {
@@ -69,6 +72,7 @@ public class LicensesServiceTests extends ElasticsearchIntegrationTest {
     @Before
     public void beforeTest() throws Exception {
         final CountDownLatch latch = new CountDownLatch(1);
+        // todo: fix with awaitBusy
         masterClusterService().submitStateUpdateTask("delete licensing metadata", new ProcessedClusterStateUpdateTask() {
             @Override
             public void clusterStateProcessed(String source, ClusterState oldState, ClusterState newState) {
@@ -121,14 +125,14 @@ public class LicensesServiceTests extends ElasticsearchIntegrationTest {
 
         assertTrue(LicensesStatus.VALID == licensesManagerService.checkLicenses(licenses));
 
-        ESLicense esLicense = Utils.reduceAndMap(licenses).get(TestUtils.SHIELD);
+        ESLicense esLicense = ESLicenses.reduceAndMap(licenses).get(TestUtils.SHIELD);
 
         final ESLicense tamperedLicense = ESLicense.builder()
                 .fromLicense(esLicense)
                 .expiryDate(esLicense.expiryDate() + 10 * 24 * 60 * 60 * 1000l)
                 .feature(TestUtils.SHIELD)
                 .issuer("elasticsqearch")
-                .build();
+                .verifyAndBuild();
 
         assertTrue(LicensesStatus.INVALID == licensesManagerService.checkLicenses(Collections.singleton(tamperedLicense)));
     }
@@ -146,6 +150,7 @@ public class LicensesServiceTests extends ElasticsearchIntegrationTest {
         LicensesManagerService licensesManagerService = masterLicensesManagerService();
         ESLicenseManager esLicenseManager = ((LicensesService) licensesManagerService).getEsLicenseManager();
         final CountDownLatch latch1 = new CountDownLatch(1);
+        // todo: fix with awaitBusy
         licensesManagerService.registerLicenses(new LicensesService.PutLicenseRequestHolder(new PutLicenseRequest().licenses(licenses), "test"), new ActionListener<ClusterStateUpdateResponse>() {
             @Override
             public void onResponse(ClusterStateUpdateResponse clusterStateUpdateResponse) {
@@ -173,6 +178,7 @@ public class LicensesServiceTests extends ElasticsearchIntegrationTest {
         licenseOutput = TestUtils.runLicenseGenerationTool(licenseString, pubKeyPath, priKeyPath);
         List<ESLicense> licenses2 = ESLicenses.fromSource(licenseOutput);
         final CountDownLatch latch2 = new CountDownLatch(1);
+        // todo: fix with awaitBusy
         licensesManagerService.registerLicenses(new LicensesService.PutLicenseRequestHolder(new PutLicenseRequest().licenses(licenses2), "test"), new ActionListener<ClusterStateUpdateResponse>() {
             @Override
             public void onResponse(ClusterStateUpdateResponse clusterStateUpdateResponse) {
@@ -197,7 +203,8 @@ public class LicensesServiceTests extends ElasticsearchIntegrationTest {
     public void testTrialLicenseGeneration() throws Exception {
         LicensesClientService clientService = licensesClientService();
         final CountDownLatch latch = new CountDownLatch(1);
-        clientService.register("shield", new LicensesService.TrialLicenseOptions(10, 100), new LicensesClientService.Listener() {
+        // todo: fix with awaitBusy
+        clientService.register("shield", new LicensesService.TrialLicenseOptions(TimeValue.timeValueHours(10), 100), new LicensesClientService.Listener() {
             @Override
             public void onEnabled() {
                 logger.info("got onEnabled from LicensesClientService");
@@ -211,8 +218,6 @@ public class LicensesServiceTests extends ElasticsearchIntegrationTest {
         });
         logger.info("waiting for onEnabled");
         latch.await();
-        final LicensesMetaData metaData = clusterService().state().metaData().custom(LicensesMetaData.TYPE);
-        assertTrue(metaData.getEncodedTrialLicenses().size() == 1);
     }
 
     @Test
@@ -262,7 +267,7 @@ public class LicensesServiceTests extends ElasticsearchIntegrationTest {
         // feature should be onEnabled
 
         LicensesClientService clientService = licensesClientService();
-        LicensesManagerService managerService = licensesManagerService();
+        final LicensesManagerService managerService = licensesManagerService();
         LicensesManagerService masterLicensesManagerService = masterLicensesManagerService();
         final TestLicenseClientListener testLicenseClientListener = new TestLicenseClientListener(false);
         clientService.register("shield", null, testLicenseClientListener);
@@ -284,6 +289,7 @@ public class LicensesServiceTests extends ElasticsearchIntegrationTest {
         List<ESLicense> licenses = ESLicenses.fromSource(licenseOutput);
 
         final CountDownLatch latch1 = new CountDownLatch(1);
+        // todo: fix with awaitBusy
         masterLicensesManagerService.registerLicenses(new LicensesService.PutLicenseRequestHolder(new PutLicenseRequest().licenses(licenses), "test"), new ActionListener<ClusterStateUpdateResponse>() {
             @Override
             public void onResponse(ClusterStateUpdateResponse clusterStateUpdateResponse) {
@@ -301,16 +307,19 @@ public class LicensesServiceTests extends ElasticsearchIntegrationTest {
         latch1.await();
 
         logger.info("waiting for onEnabled");
-        while (!testLicenseClientListener.processed.get()) {
+        assertThat(awaitBusy(new Predicate<Object>() {
-        }
+            @Override
-
+            public boolean apply(Object o) {
-        assertTrue(managerService.enabledFeatures().contains("shield"));
+                return managerService.enabledFeatures().contains("shield");
+            }
+        }, 1, TimeUnit.MINUTES), equalTo(true));
 
     }
 
     @Test
     public void testFeatureWithoutLicense() throws Exception {
         LicensesClientService clientService = licensesClientService();
+        // todo: fix with awaitBusy
         clientService.register("marvel", null, new LicensesClientService.Listener() {
             @Override
             public void onEnabled() {
@@ -345,11 +354,6 @@ public class LicensesServiceTests extends ElasticsearchIntegrationTest {
         return internalCluster().getInstance(LicensesClientService.class, node);
     }
 
-    private LicensesService licensesService() {
-        final InternalTestCluster clients = internalCluster();
-        return clients.getInstance(LicensesService.class, clients.getMasterName());
-    }
-
     private static ClusterService masterClusterService() {
         final InternalTestCluster clients = internalCluster();
         return clients.getInstance(ClusterService.class, clients.getMasterName());

src/test/java/org/elasticsearch/license/plugin/TestConsumerPlugin.java

@@ -7,13 +7,19 @@ package org.elasticsearch.license.plugin;
 
 import org.elasticsearch.common.collect.ImmutableSet;
 import org.elasticsearch.common.component.LifecycleComponent;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.plugins.AbstractPlugin;
 
 import java.util.Collection;
 
 public class TestConsumerPlugin extends AbstractPlugin {
 
-    public TestConsumerPlugin() {
+    private final Settings settings;
+
+    @Inject
+    public TestConsumerPlugin(Settings settings) {
+    this.settings = settings;
     }
 
     @Override

src/test/java/org/elasticsearch/license/plugin/TestPluginService.java

@@ -10,6 +10,7 @@ import org.elasticsearch.common.component.AbstractLifecycleComponent;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.inject.Singleton;
 import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.TimeValue;
 import org.elasticsearch.license.plugin.core.LicensesClientService;
 import org.elasticsearch.license.plugin.core.LicensesService;
 
@@ -26,7 +27,7 @@ public class TestPluginService extends AbstractLifecycleComponent<TestPluginServ
 
     // specify the trial license spec for the feature
     // example: 30 day trial on 1000 nodes
-    final LicensesService.TrialLicenseOptions trialLicenseOptions = new LicensesService.TrialLicenseOptions(30, 1000);
+    final LicensesService.TrialLicenseOptions trialLicenseOptions;
 
     private AtomicBoolean enabled = new AtomicBoolean(false);
 
@@ -34,6 +35,13 @@ public class TestPluginService extends AbstractLifecycleComponent<TestPluginServ
     public TestPluginService(Settings settings, LicensesClientService licensesClientService) {
         super(settings);
         this.licensesClientService = licensesClientService;
+        int durationInSec = settings.getAsInt("test_consumer_plugin.trial_license_duration_in_seconds", -1);
+        logger.info("Trial license Duration in seconds: " + durationInSec);
+        if (durationInSec == -1) {
+            this.trialLicenseOptions = null;
+        } else {
+            this.trialLicenseOptions = new LicensesService.TrialLicenseOptions(TimeValue.timeValueSeconds(durationInSec), 1000);
+        }
     }
 
     // check if feature is enabled