feat: Auto-detect Elasticsearch version and improve partitioned table handling

Author: mertbilgic

README.md

@@ -176,7 +176,7 @@ This setup ensures continuous data synchronization and minimal downtime in captu
 | `elasticsearch.compressionEnabled`                                                           |       bool        |              no               |  false  | Enable compression for large messages                                                                 | Useful if message sizes are large, but may increase CPU usage.                                                                                                           |
 | `elasticsearch.disableDiscoverNodesOnStart`                                                  |       bool        |              no               |  false  | Disable node discovery on client initialization                                                       | Skips node discovery when the client starts.                                                                                                                             |
 | `elasticsearch.discoverNodesInterval`                                                        |   time.Duration   |              no               |  5 min  | Periodic node discovery interval                                                                      | Specify in a human-readable format, e.g., `5m` for 5 minutes.                                                                                                            |
-| `elasticsearch.version`                                                                      |      string       |              no               | 6.8.0   | Elasticsearch version to determine compatibility features                                             | Used to handle version-specific behaviors, such as `_type` parameter support (removed in ES 8.0+).                                                                       |
+| `elasticsearch.version`                                                                      |      string       |              no               |    -    | Elasticsearch version to determine compatibility features                                             | Used to handle version-specific behaviors, such as `_type` parameter support (removed in ES 8.0+). If not specified, version is automatically detected from the cluster. |
 
 ### API
 
@@ -216,7 +216,7 @@ The connector supports different versions of Elasticsearch through the `elastics
 | Below 8.0             | `_type` parameter is included in the index requests     |
 | 8.0 and above         | `_type` parameter is automatically omitted              |
 
-If no version is specified, the connector defaults to "7.0.0" behavior (including `_type` parameter).
+If no version is specified, the connector will automatically detect the Elasticsearch cluster version by querying the Info API after connection. This eliminates the need to manually configure the version.
 
 ### Breaking Changes
 

config/config.go

@@ -55,8 +55,4 @@ func (c *Config) SetDefault() {
 		duration := 5 * time.Minute
 		c.Elasticsearch.DiscoverNodesInterval = &duration
 	}
-
-	if c.Elasticsearch.Version == "" {
-		c.Elasticsearch.Version = "7.0.0"
-	}
 }

connector.go

@@ -139,23 +139,23 @@ func (c *connector) listener(ctx *replication.ListenerContext) {
 func (c *connector) resolveTableToIndexName(fullTableName, tableNamespace, tableName string) string {
 	tableIndexMapping := c.cfg.Elasticsearch.TableIndexMapping
 	if len(tableIndexMapping) == 0 {
-		return fullTableName
+		return ""
 	}
 
 	if indexName, exists := tableIndexMapping[fullTableName]; exists {
 		return indexName
 	}
 
-	parentTableName := c.getParentTableName(fullTableName, tableNamespace, tableName)
-	if parentTableName != "" {
-		if indexName, exists := tableIndexMapping[parentTableName]; exists {
+	if t, ok := timescaledb.HyperTables.Load(fullTableName); ok {
+		parentName := t.(string)
+		if indexName, exists := tableIndexMapping[parentName]; exists {
 			return indexName
 		}
 	}
 
-	if t, ok := timescaledb.HyperTables.Load(fullTableName); ok {
-		parentName := t.(string)
-		if indexName, exists := tableIndexMapping[parentName]; exists {
+	parentTableName := c.getParentTableName(fullTableName, tableNamespace, tableName)
+	if parentTableName != "" {
+		if indexName, exists := tableIndexMapping[parentTableName]; exists {
 			return indexName
 		}
 	}

elasticsearch/client/client.go

@@ -1,7 +1,10 @@
 package client
 
 import (
+	"encoding/json"
+
 	"github.com/Trendyol/go-pq-cdc-elasticsearch/config"
+	"github.com/Trendyol/go-pq-cdc/logger"
 	"github.com/elastic/go-elasticsearch/v7"
 	"github.com/go-playground/errors"
 )
@@ -31,5 +34,41 @@ func NewClient(config *config.Config) (*elasticsearch.Client, error) {
 		return nil, errors.New("unauthorized")
 	}
 
+	if config.Elasticsearch.Version == "" {
+		version, err := detectElasticsearchVersion(client)
+		if err != nil {
+			logger.Warn("elasticsearch version detection failed", "error", err, "fallback_version", "7.0.0", "hint", "specify 'elasticsearch.version' in config to set manually")
+			config.Elasticsearch.Version = "7.0.0"
+		} else {
+			logger.Info("elasticsearch version detected", "version", version)
+			config.Elasticsearch.Version = version
+		}
+	}
+
 	return client, nil
 }
+
+func detectElasticsearchVersion(client *elasticsearch.Client) (string, error) {
+	info, err := client.Info()
+	if err != nil {
+		return "", err
+	}
+
+	var response map[string]interface{}
+	if err := json.NewDecoder(info.Body).Decode(&response); err != nil {
+		return "", err
+	}
+	defer info.Body.Close()
+
+	version, ok := response["version"].(map[string]interface{})
+	if !ok {
+		return "", errors.New("version info not found in Elasticsearch response")
+	}
+
+	number, ok := version["number"].(string)
+	if !ok {
+		return "", errors.New("version number not found in Elasticsearch response")
+	}
+
+	return number, nil
+}