Fix bwc in GeoDistanceQuery serialization

Restores backward compatibility in GeoDistanceQueryBuilder serialization
between 6.x and 5.6 after removal of optimize_bbox parameter.

Closes elastic#29301

qa/rolling-upgrade/build.gradle

@@ -43,6 +43,7 @@ for (Version version : bwcVersions.wireCompatible) {
     numNodes = 2
     clusterName = 'rolling-upgrade'
     setting 'repositories.url.allowed_urls', 'http://snapshot.test*'
+    setting 'node.attr.gen', 'old'
     if (version.onOrAfter('5.3.0')) {
       setting 'http.content_type.required', 'true'
     }
@@ -64,6 +65,7 @@ for (Version version : bwcVersions.wireCompatible) {
      * just stopped's data directory. */
     dataDir = { nodeNumber -> oldClusterTest.nodes[1].dataDir }
     setting 'repositories.url.allowed_urls', 'http://snapshot.test*'
+    setting 'node.attr.gen', 'new'
   }
 
   Task mixedClusterTestRunner = tasks.getByName("${baseName}#mixedClusterTestRunner")
@@ -83,6 +85,7 @@ for (Version version : bwcVersions.wireCompatible) {
      * just stopped's data directory. */
     dataDir = { nodeNumber -> oldClusterTest.nodes[0].dataDir}
     setting 'repositories.url.allowed_urls', 'http://snapshot.test*'
+    setting 'node.attr.gen', 'new'
   }
 
   Task upgradedClusterTestRunner = tasks.getByName("${baseName}#upgradedClusterTestRunner")

qa/rolling-upgrade/src/test/java/org/elasticsearch/upgrades/RecoveryIT.java

@@ -266,4 +266,53 @@ public void testRelocationWithConcurrentIndexing() throws Exception {
         }
     }
 
+    public void testSearchGeoPoints() throws Exception {
+        final String index = "geo_index";
+        if (clusterType == CLUSTER_TYPE.OLD) {
+            Settings.Builder settings = Settings.builder()
+                .put(IndexMetaData.INDEX_NUMBER_OF_SHARDS_SETTING.getKey(), 1)
+                .put(IndexMetaData.INDEX_NUMBER_OF_REPLICAS_SETTING.getKey(), 1)
+                // if the node with the replica is the first to be restarted, while a replica is still recovering
+                // then delayed allocation will kick in. When the node comes back, the master will search for a copy
+                // but the recovering copy will be seen as invalid and the cluster health won't return to GREEN
+                // before timing out
+                .put(INDEX_DELAYED_NODE_LEFT_TIMEOUT_SETTING.getKey(), "100ms");
+            createIndex(index, settings.build(), "\"doc\": {\"properties\": {\"location\": {\"type\": \"geo_point\"}}}");
+            assertOK(client().performRequest("PUT", index + "/doc/1", emptyMap(),
+                new StringEntity("{\"location\": { \"lat\": 42.358056, \"lon\": -71.063611 }, \"foo\": \"bar\"}", ContentType.APPLICATION_JSON)));
+            assertOK(client().performRequest("PUT", index + "/doc/2", emptyMap(),
+                new StringEntity("{\"location\": { \"lat\": 37.783333, \"lon\": -122.416667 }, \"foo\": \"bar\"}", ContentType.APPLICATION_JSON)));
+            assertOK(client().performRequest("POST", index + "/_refresh"));
+            ensureGreen(index);
+        } else if (clusterType == CLUSTER_TYPE.MIXED) {
+            ensureGreen(index);
+            String requestBody = "{\n" +
+                "  \"query\": {\n" +
+                "    \"bool\": {\n" +
+                "      \"should\": [\n" +
+                "        {\n" +
+                "          \"geo_distance\": {\n" +
+                "            \"distance\": \"1000km\",\n" +
+                "            \"location\": {\n" +
+                "              \"lat\": 40,\n" +
+                "              \"lon\": -70\n" +
+                "            }\n" +
+                "          }\n" +
+                "        },\n" +
+                "        {\"match_all\": {}}\n" +
+                "      ]\n" +
+                "    }\n" +
+                "  }\n" +
+                "}";
+
+            // we need to make sure that requests are routed from a new node to the old node so we are sending the request a few times
+            for (int i = 0; i < 10; i++) {
+                Response response = client().performRequest("GET", index + "/_search",
+                    Collections.singletonMap("preference", "_only_nodes:gen:old"), // Make sure we only send this request to old nodes
+                    new StringEntity(requestBody, ContentType.APPLICATION_JSON));
+                assertOK(response);
+            }
+        }
+    }
+
 }

server/src/main/java/org/elasticsearch/index/query/GeoDistanceQueryBuilder.java

@@ -24,6 +24,7 @@
 import org.apache.lucene.search.IndexOrDocValuesQuery;
 import org.apache.lucene.search.MatchNoDocsQuery;
 import org.apache.lucene.search.Query;
+import org.elasticsearch.Version;
 import org.elasticsearch.common.ParseField;
 import org.elasticsearch.common.ParsingException;
 import org.elasticsearch.common.Strings;
@@ -97,6 +98,10 @@ public GeoDistanceQueryBuilder(StreamInput in) throws IOException {
         distance = in.readDouble();
         validationMethod = GeoValidationMethod.readFromStream(in);
         center = in.readGeoPoint();
+        if (in.getVersion().before(Version.V_6_0_0_alpha1)) {
+            // optimize bounding box was removed in 6.0
+            in.readOptionalString();
+        }
         geoDistance = GeoDistance.readFromStream(in);
         ignoreUnmapped = in.readBoolean();
     }
@@ -107,6 +112,10 @@ protected void doWriteTo(StreamOutput out) throws IOException {
         out.writeDouble(distance);
         validationMethod.writeTo(out);
         out.writeGeoPoint(center);
+        if (out.getVersion().before(Version.V_6_0_0_alpha1)) {
+            // optimize bounding box was removed in 6.0
+            out.writeOptionalString(null);
+        }
         geoDistance.writeTo(out);
         out.writeBoolean(ignoreUnmapped);
     }