UnicastHostsProvider should use version.minimumCompatibilityVersion()

The UnicastHostsProvider implementation creates DiscoveryNodes that are used as an initial seed for unicast based discovery. At the moment it uses Version.CURRENT for those DiscoveryNode object, which confuses the backwards compatibility layer to think this nodes are of the latest version. This causes new nodes to fail to join old nodes as the ping serialization goes wrong. Instead we should use version.minimumCompatibilityVersion().

Closes #47.

(cherry picked from commit 188179f)

Author: dadoonet

src/main/java/org/elasticsearch/discovery/azure/AzureDiscovery.java

@@ -19,6 +19,7 @@
 
 package org.elasticsearch.discovery.azure;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.cloud.azure.AzureComputeService;
 import org.elasticsearch.cluster.ClusterName;
 import org.elasticsearch.cluster.ClusterService;
@@ -49,7 +50,8 @@ public class AzureDiscovery extends ZenDiscovery {
     public AzureDiscovery(Settings settings, ClusterName clusterName, ThreadPool threadPool, TransportService transportService,
                           ClusterService clusterService, NodeSettingsService nodeSettingsService, ZenPingService pingService,
                           DiscoveryNodeService discoveryNodeService, AzureComputeService azureService, NetworkService networkService,
-                          DiscoverySettings discoverySettings, ElectMasterService electMasterService, DynamicSettings dynamicSettings) {
+                          DiscoverySettings discoverySettings, ElectMasterService electMasterService, DynamicSettings dynamicSettings,
+                          Version version) {
         super(settings, clusterName, threadPool, transportService, clusterService, nodeSettingsService,
                 discoveryNodeService, pingService, electMasterService, discoverySettings, dynamicSettings);
         if (settings.getAsBoolean("cloud.enabled", true)) {
@@ -65,7 +67,7 @@ public AzureDiscovery(Settings settings, ClusterName clusterName, ThreadPool thr
             if (unicastZenPing != null) {
                 // update the unicast zen ping to add cloud hosts provider
                 // and, while we are at it, use only it and not the multicast for example
-                unicastZenPing.addHostsProvider(new AzureUnicastHostsProvider(settings, azureService, transportService, networkService));
+                unicastZenPing.addHostsProvider(new AzureUnicastHostsProvider(settings, azureService, transportService, networkService, version));
                 pingService.zenPings(ImmutableList.of(unicastZenPing));
             } else {
                 logger.warn("failed to apply azure unicast discovery, no unicast ping found");

src/main/java/org/elasticsearch/discovery/azure/AzureUnicastHostsProvider.java

@@ -53,6 +53,7 @@ public static enum HostType {
     private final AzureComputeService azureComputeService;
     private TransportService transportService;
     private NetworkService networkService;
+    private final Version version;
 
     private final TimeValue refreshInterval;
     private long lastRefresh;
@@ -62,11 +63,13 @@ public static enum HostType {
     @Inject
     public AzureUnicastHostsProvider(Settings settings, AzureComputeService azureComputeService,
                                    TransportService transportService,
-                                   NetworkService networkService) {
+                                   NetworkService networkService,
+                                   Version version) {
         super(settings);
         this.azureComputeService = azureComputeService;
         this.transportService = transportService;
         this.networkService = networkService;
+        this.version = version;
 
         this.refreshInterval = componentSettings.getAsTime(AzureComputeService.Fields.REFRESH,
                 settings.getAsTime("cloud.azure." + AzureComputeService.Fields.REFRESH, TimeValue.timeValueSeconds(0)));
@@ -139,7 +142,7 @@ public List<DiscoveryNode> buildDynamicNodes() {
                     TransportAddress[] addresses = transportService.addressesFromString(networkAddress);
                     // we only limit to 1 addresses, makes no sense to ping 100 ports
                     logger.trace("adding {}, transport_address {}", networkAddress, addresses[0]);
-                    cachedDiscoNodes.add(new DiscoveryNode("#cloud-" + instance.getName(), addresses[0], Version.CURRENT));
+                    cachedDiscoNodes.add(new DiscoveryNode("#cloud-" + instance.getName(), addresses[0], version.minimumCompatibilityVersion()));
                 }
 
             }