DATAES-435 - Report version mismatch if used with older ElasticSearch…

pom.xml

@@ -292,6 +292,23 @@
 	</dependencies>
 
 	<build>
+		<resources>
+			<resource>
+				<directory>src/main/resources</directory>
+				<filtering>true</filtering>
+				<includes>
+					<include>**/versions.properties</include>
+				</includes>
+			</resource>
+			<resource>
+				<directory>src/main/resources</directory>
+				<filtering>false</filtering>
+				<excludes>
+					<exclude>**/versions.properties</exclude>
+				</excludes>
+			</resource>
+		</resources>
+
 		<plugins>
 			<!--
 			please do not remove this configuration for surefire - we need that to avoid issue with jar hell

src/main/java/org/springframework/data/elasticsearch/core/AbstractElasticsearchTemplate.java

@@ -33,6 +33,7 @@
 import org.springframework.data.elasticsearch.core.query.MoreLikeThisQuery;
 import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
 import org.springframework.data.elasticsearch.core.query.Query;
+import org.springframework.data.elasticsearch.support.VersionInfo;
 import org.springframework.data.mapping.callback.EntityCallbacks;
 import org.springframework.data.util.CloseableIterator;
 import org.springframework.data.util.Streamable;
@@ -59,6 +60,8 @@ protected void initialize(ElasticsearchConverter elasticsearchConverter) {
 
 		this.elasticsearchConverter = elasticsearchConverter;
 		requestFactory = new RequestFactory(elasticsearchConverter);
+
+		VersionInfo.logVersions(getClusterVersion());
 	}
 
 	protected ElasticsearchConverter createElasticsearchConverter() {
@@ -288,14 +291,14 @@ public <T> SearchHits<T> search(Query query, Class<T> clazz) {
 	/*
 	 * internal use only, not for public API
 	 */
-	abstract protected <T> SearchScrollHits<T> searchScrollStart(long scrollTimeInMillis, Query query,
-			Class<T> clazz, IndexCoordinates index);
+	abstract protected <T> SearchScrollHits<T> searchScrollStart(long scrollTimeInMillis, Query query, Class<T> clazz,
+			IndexCoordinates index);
 
 	/*
 	 * internal use only, not for public API
 	 */
-	abstract protected <T> SearchScrollHits<T> searchScrollContinue(@Nullable String scrollId,
-			long scrollTimeInMillis, Class<T> clazz);
+	abstract protected <T> SearchScrollHits<T> searchScrollContinue(@Nullable String scrollId, long scrollTimeInMillis,
+			Class<T> clazz);
 
 	/*
 	 * internal use only, not for public API
@@ -417,6 +420,15 @@ private <T> IndexQuery getIndexQuery(T entity) {
 				.withObject(entity) //
 				.build();
 	}
+
+	/**
+	 * tries to extract the version of the Elasticsearch cluster
+	 * 
+	 * @return the version as string if it can be retrieved
+	 */
+	@Nullable
+	abstract protected String getClusterVersion();
+
 	// endregion
 
 	// region callbacks

src/main/java/org/springframework/data/elasticsearch/core/ElasticsearchRestTemplate.java

@@ -270,7 +270,8 @@ public <T> SearchScrollHits<T> searchScrollStart(long scrollTimeInMillis, Query
 	}
 
 	@Override
-	public <T> SearchScrollHits<T> searchScrollContinue(@Nullable String scrollId, long scrollTimeInMillis, Class<T> clazz) {
+	public <T> SearchScrollHits<T> searchScrollContinue(@Nullable String scrollId, long scrollTimeInMillis,
+			Class<T> clazz) {
 
 		SearchScrollRequest request = new SearchScrollRequest(scrollId);
 		request.scroll(TimeValue.timeValueMillis(scrollTimeInMillis));
@@ -355,4 +356,14 @@ private RuntimeException translateException(Exception exception) {
 	}
 
 	// endregion
+
+	// region helper methods
+	@Override
+	protected String getClusterVersion() {
+		try {
+			return execute(client -> client.info(RequestOptions.DEFAULT)).getVersion().getNumber();
+		} catch (Exception ignored) {}
+		return null;
+	}
+	// endregion
 }

src/main/java/org/springframework/data/elasticsearch/core/ElasticsearchTemplate.java

@@ -17,7 +17,11 @@
 
 import java.util.List;
 
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionFuture;
+import org.elasticsearch.action.admin.cluster.node.info.NodesInfoAction;
+import org.elasticsearch.action.admin.cluster.node.info.NodesInfoRequestBuilder;
+import org.elasticsearch.action.admin.cluster.node.info.NodesInfoResponse;
 import org.elasticsearch.action.bulk.BulkRequestBuilder;
 import org.elasticsearch.action.get.GetRequestBuilder;
 import org.elasticsearch.action.get.GetResponse;
@@ -320,4 +324,22 @@ private SearchResponse getSearchResponseWithTimeout(ActionFuture<SearchResponse>
 		return searchTimeout == null ? response.actionGet() : response.actionGet(searchTimeout);
 	}
 	// endregion
+
+
+	//region helper methods
+	@Override
+	protected String getClusterVersion() {
+
+		try {
+			NodesInfoResponse nodesInfoResponse = client.admin().cluster().nodesInfo(
+					new NodesInfoRequestBuilder(client, NodesInfoAction.INSTANCE).request()
+			).actionGet();
+			if (!nodesInfoResponse.getNodes().isEmpty()) {
+				return nodesInfoResponse.getNodes().get(0).getVersion().toString();
+			}
+		} catch (Exception ignored) {
+		}
+		return null;
+	}
+	//endregion
 }

src/main/java/org/springframework/data/elasticsearch/core/ReactiveElasticsearchTemplate.java

@@ -82,6 +82,7 @@
 import org.springframework.data.elasticsearch.core.query.Query;
 import org.springframework.data.elasticsearch.core.query.StringQuery;
 import org.springframework.data.elasticsearch.core.query.UpdateQuery;
+import org.springframework.data.elasticsearch.support.VersionInfo;
 import org.springframework.data.mapping.callback.ReactiveEntityCallbacks;
 import org.springframework.data.mapping.context.MappingContext;
 import org.springframework.http.HttpStatus;
@@ -131,6 +132,15 @@ public ReactiveElasticsearchTemplate(ReactiveElasticsearchClient client, Elastic
 		this.exceptionTranslator = new ElasticsearchExceptionTranslator();
 		this.operations = new EntityOperations(this.mappingContext);
 		this.requestFactory = new RequestFactory(converter);
+
+		logVersions();
+	}
+
+	private void logVersions() {
+		getClusterVersion() //
+				.doOnSuccess(VersionInfo::logVersions) //
+				.doOnError(e -> VersionInfo.logVersions(null)) //
+				.subscribe();
 	}
 
 	@Override
@@ -157,6 +167,7 @@ public void setEntityCallbacks(ReactiveEntityCallbacks entityCallbacks) {
 
 		this.entityCallbacks = entityCallbacks;
 	}
+
 	// endregion
 
 	// region DocumentOperations
@@ -775,6 +786,15 @@ protected SearchRequest prepareSearchRequest(SearchRequest request) {
 	// endregion
 
 	// region Helper methods
+	protected Mono<String> getClusterVersion() {
+		try {
+			return Mono.from(execute(client -> client.info())).map(mainResponse -> mainResponse.getVersion().toString());
+		} catch (Exception ignored) {}
+		return Mono.empty();
+	}
+
+	// endregion
+
 	// Property Setters / Getters
 
 	/**

src/main/java/org/springframework/data/elasticsearch/support/VersionInfo.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2020 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.elasticsearch.support;
+
+import java.io.InputStream;
+import java.util.Properties;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.elasticsearch.Version;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.lang.Nullable;
+
+/**
+ * This class is used to log the versions of Spring Data Elasticsearch, the Elasticsearch client libs used to built, the
+ * Elasticsearch client libs currently used and of the Elasticsearch cluster. If differences greater than a patchlevel
+ * are detected, these are logged as warnings.
+ *
+ * @author Peter-Josef Meisch
+ * @since 4.0
+ */
+public final class VersionInfo {
+
+	private static final Logger LOG = LoggerFactory.getLogger(VersionInfo.class);
+	private static final AtomicBoolean initialized = new AtomicBoolean(false);
+	private static String VERSION_PROPERTIES = "versions.properties";
+
+	/**
+	 * logs the relevant version info the first time it is called. Does nothing after the first call
+	 * 
+	 * @param clusterVersion the version of the cluster
+	 */
+	public static void logVersions(@Nullable String clusterVersion) {
+		if (!initialized.getAndSet(true)) {
+			try {
+				InputStream resource = VersionInfo.class.getClassLoader().getResourceAsStream(VERSION_PROPERTIES);
+				if (resource != null) {
+					Properties properties = new Properties();
+					properties.load(resource);
+
+					String versionSpringDataElasticsearch = properties.getProperty("version.spring-data-elasticsearch");
+					Version versionESBuilt = Version.fromString(properties.getProperty("version.elasticsearch-client"));
+					Version versionESUsed = Version.CURRENT;
+					Version versionESCluster = clusterVersion != null ? Version.fromString(clusterVersion) : null;
+
+					LOG.info("Version Spring Data Elasticsearch: {}", versionSpringDataElasticsearch.toString());
+					LOG.info("Version Elasticsearch Client in build: {}", versionESBuilt.toString());
+					LOG.info("Version Elasticsearch Client used: {}", versionESUsed.toString());
+
+					if (differInMajorOrMinor(versionESBuilt, versionESUsed)) {
+						LOG.warn("Version mismatch in between Elasticsearch Clients build/use: {} - {}", versionESBuilt,
+								versionESUsed);
+					}
+
+					if (versionESCluster != null) {
+						LOG.info("Version Elasticsearch cluster: {}", versionESCluster.toString());
+
+						if (differInMajorOrMinor(versionESUsed, versionESCluster)) {
+							LOG.warn("Version mismatch in between Elasticsearch Client and Cluster: {} - {}", versionESUsed,
+									versionESCluster);
+						}
+					}
+				} else {
+					LOG.warn("cannot load {}", VERSION_PROPERTIES);
+				}
+			} catch (Exception e) {
+				LOG.warn("Could not log version info: {} - {}", e.getClass().getSimpleName(), e.getMessage());
+			}
+
+		}
+	}
+
+	private static boolean differInMajorOrMinor(Version version1, Version version2) {
+		return version1.major != version2.major || version1.minor != version2.minor;
+	}
+
+	private VersionInfo() {}
+}

src/main/resources/versions.properties

@@ -0,0 +1,2 @@
+version.spring-data-elasticsearch=${project.version}
+version.elasticsearch-client=${elasticsearch}

src/test/resources/logback.xml

@@ -7,6 +7,7 @@
 		</encoder>
 	</appender>
 
+	<logger name="org.springframework.data.elasticsearch.support.VersionInfo" level="info"/>
 	<logger name="org.springframework.data.elasticsearch.junit" level="debug"/>
 	<logger name="org.springframework.data.elasticsearch.junit.junit4.ElasticsearchVersionRule" level="info"/>
 	<logger name="org.springframework.data.elasticsearch.client.WIRE" level="info"/>