class NodeVersionAllocationDecider extends AllocationDecider {

    public static final String NAME = "node_version";

    @Override
    public Decision canAllocate(ShardRouting shardRouting, RoutingNode node, RoutingAllocation allocation) {<FILL_FUNCTION_BODY>}

    private Decision isVersionCompatibleRelocatePrimary(final RoutingNodes routingNodes, final String sourceNodeId,
                                                        final RoutingNode target, final RoutingAllocation allocation) {
        final RoutingNode source = routingNodes.node(sourceNodeId);
        if (target.node().getVersion().onOrAfterMajorMinor(source.node().getVersion())) {
            return allocation.decision(Decision.YES, NAME,
                "can relocate primary shard from a node with version [%s] to a node with equal-or-newer version [%s]",
                source.node().getVersion(), target.node().getVersion());
        } else {
            return allocation.decision(Decision.NO, NAME,
                "cannot relocate primary shard from a node with version [%s] to a node with older version [%s]",
                source.node().getVersion(), target.node().getVersion());
        }
    }

    private Decision isVersionCompatibleAllocatingReplica(final RoutingNodes routingNodes, final String sourceNodeId,
                                                          final RoutingNode target, final RoutingAllocation allocation) {
        final RoutingNode source = routingNodes.node(sourceNodeId);
        if (target.node().getVersion().onOrAfterMajorMinor(source.node().getVersion())) {
            /* we can allocate if we can recover from a node that is younger or on the same version
             * if the primary is already running on a newer version that won't work due to possible
             * differences in the lucene index format etc.*/
            return allocation.decision(Decision.YES, NAME,
                "can allocate replica shard to a node with version [%s] since this is equal-or-newer than the primary version [%s]",
                target.node().getVersion(), source.node().getVersion());
        } else {
            return allocation.decision(Decision.NO, NAME,
                "cannot allocate replica shard to a node with version [%s] since this is older than the primary version [%s]",
                target.node().getVersion(), source.node().getVersion());
        }
    }

    private Decision isVersionCompatible(SnapshotRecoverySource recoverySource, final RoutingNode target,
                                         final RoutingAllocation allocation) {
        if (target.node().getVersion().onOrAfter(recoverySource.version())) {
            /* we can allocate if we can restore from a snapshot that is older or on the same version */
            return allocation.decision(Decision.YES, NAME, "node version [%s] is the same or newer than snapshot version [%s]",
                target.node().getVersion(), recoverySource.version());
        } else {
            return allocation.decision(Decision.NO, NAME, "node version [%s] is older than the snapshot version [%s]",
                target.node().getVersion(), recoverySource.version());
        }
    }
}

        if (shardRouting.primary()) {
            if (shardRouting.currentNodeId() == null) {
                if (shardRouting.recoverySource() != null && shardRouting.recoverySource().getType() == RecoverySource.Type.SNAPSHOT) {
                    // restoring from a snapshot - check that the node can handle the version
                    return isVersionCompatible((SnapshotRecoverySource)shardRouting.recoverySource(), node, allocation);
                } else {
                    // existing or fresh primary on the node
                    return allocation.decision(Decision.YES, NAME, "the primary shard is new or already existed on the node");
                }
            } else {
                // relocating primary, only migrate to newer host
                return isVersionCompatibleRelocatePrimary(allocation.routingNodes(), shardRouting.currentNodeId(), node, allocation);
            }
        } else {
            final ShardRouting primary = allocation.routingNodes().activePrimary(shardRouting.shardId());
            // check that active primary has a newer version so that peer recovery works
            if (primary != null) {
                return isVersionCompatibleAllocatingReplica(allocation.routingNodes(), primary.currentNodeId(), node, allocation);
            } else {
                // ReplicaAfterPrimaryActiveAllocationDecider should prevent this case from occurring
                return allocation.decision(Decision.YES, NAME, "no active primary shard yet");
            }
        }
    
<methods>public org.elasticsearch.cluster.routing.allocation.decider.Decision canAllocate(org.elasticsearch.cluster.routing.ShardRouting, org.elasticsearch.cluster.routing.RoutingNode, org.elasticsearch.cluster.routing.allocation.RoutingAllocation) ,public org.elasticsearch.cluster.routing.allocation.decider.Decision canAllocate(org.elasticsearch.cluster.routing.ShardRouting, org.elasticsearch.cluster.routing.allocation.RoutingAllocation) ,public org.elasticsearch.cluster.routing.allocation.decider.Decision canAllocate(org.elasticsearch.cluster.metadata.IndexMetadata, org.elasticsearch.cluster.routing.RoutingNode, org.elasticsearch.cluster.routing.allocation.RoutingAllocation) ,public org.elasticsearch.cluster.routing.allocation.decider.Decision canForceAllocatePrimary(org.elasticsearch.cluster.routing.ShardRouting, org.elasticsearch.cluster.routing.RoutingNode, org.elasticsearch.cluster.routing.allocation.RoutingAllocation) ,public org.elasticsearch.cluster.routing.allocation.decider.Decision canRebalance(org.elasticsearch.cluster.routing.ShardRouting, org.elasticsearch.cluster.routing.allocation.RoutingAllocation) ,public org.elasticsearch.cluster.routing.allocation.decider.Decision canRebalance(org.elasticsearch.cluster.routing.allocation.RoutingAllocation) ,public org.elasticsearch.cluster.routing.allocation.decider.Decision canRemain(org.elasticsearch.cluster.routing.ShardRouting, org.elasticsearch.cluster.routing.RoutingNode, org.elasticsearch.cluster.routing.allocation.RoutingAllocation) ,public org.elasticsearch.cluster.routing.allocation.decider.Decision shouldAutoExpandToNode(org.elasticsearch.cluster.metadata.IndexMetadata, org.elasticsearch.cluster.node.DiscoveryNode, org.elasticsearch.cluster.routing.allocation.RoutingAllocation) <variables>