fix: add support for elasticsearch v8 indexing

Author: 3n0ugh

README.md

@@ -176,6 +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               |    -    | 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
 
@@ -206,6 +207,17 @@ All cdc related metrics are automatically injected. It means you don't need to d
 |-------------------|-----------------------------------|
 | 0.0.2 or higher   | 14                                |
 
+### Elasticsearch Version Compatibility
+
+The connector supports different versions of Elasticsearch through the `elasticsearch.version` configuration parameter:
+
+| Elasticsearch Version | Type Parameter Behavior                                 |
+|-----------------------|--------------------------------------------------------|
+| 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 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
 
 | Date taking effect | Version | Change | How to check |

config/config.go

@@ -15,6 +15,7 @@ type Elasticsearch struct {
 	MaxIdleConnDuration         *time.Duration    `yaml:"maxIdleConnDuration"`
 	DiscoverNodesInterval       *time.Duration    `yaml:"discoverNodesInterval"`
 	TypeName                    string            `yaml:"typeName"`
+	Version                     string            `yaml:"version"`
 	URLs                        []string          `yaml:"urls"`
 	BatchSizeLimit              int               `yaml:"batchSizeLimit"`
 	BatchTickerDuration         time.Duration     `yaml:"batchTickerDuration"`

connector.go

@@ -106,7 +106,10 @@ func (c *connector) listener(ctx *replication.ListenerContext) {
 		return
 	}
 
-	if !c.processMessage(msg) {
+	fullTableName := c.getFullTableName(msg.TableNamespace, msg.TableName)
+
+	indexName := c.resolveTableToIndexName(fullTableName, msg.TableNamespace, msg.TableName)
+	if indexName == "" {
 		if err := ctx.Ack(); err != nil {
 			logger.Error("ack", "error", err)
 		}
@@ -126,32 +129,38 @@ func (c *connector) listener(ctx *replication.ListenerContext) {
 		chunks := slices.ChunkWithSize[elasticsearch.Action](actions, batchSizeLimit)
 		lastChunkIndex := len(chunks) - 1
 		for idx, chunk := range chunks {
-			c.bulk.AddActions(ctx, msg.EventTime, chunk, msg.TableNamespace, msg.TableName, idx == lastChunkIndex)
+			c.bulk.AddActions(ctx, msg.EventTime, chunk, indexName, idx == lastChunkIndex)
 		}
 	} else {
-		c.bulk.AddActions(ctx, msg.EventTime, actions, msg.TableNamespace, msg.TableName, true)
+		c.bulk.AddActions(ctx, msg.EventTime, actions, indexName, true)
 	}
 }
 
-func (c *connector) processMessage(msg Message) bool {
-	if len(c.cfg.Elasticsearch.TableIndexMapping) == 0 {
-		return true
+func (c *connector) resolveTableToIndexName(fullTableName, tableNamespace, tableName string) string {
+	tableIndexMapping := c.cfg.Elasticsearch.TableIndexMapping
+	if len(tableIndexMapping) == 0 {
+		return ""
 	}
 
-	fullTableName := c.getFullTableName(msg.TableNamespace, msg.TableName)
+	if indexName, exists := tableIndexMapping[fullTableName]; exists {
+		return indexName
+	}
 
-	if _, exists := c.cfg.Elasticsearch.TableIndexMapping[fullTableName]; exists {
-		return true
+	if t, ok := timescaledb.HyperTables.Load(fullTableName); ok {
+		parentName := t.(string)
+		if indexName, exists := tableIndexMapping[parentName]; exists {
+			return indexName
+		}
 	}
 
-	t, ok := timescaledb.HyperTables.Load(fullTableName)
-	if ok {
-		_, exists := c.cfg.Elasticsearch.TableIndexMapping[t.(string)]
-		return exists
+	parentTableName := c.getParentTableName(fullTableName, tableNamespace, tableName)
+	if parentTableName != "" {
+		if indexName, exists := tableIndexMapping[parentTableName]; exists {
+			return indexName
+		}
 	}
 
-	parentTableName := c.getParentTableName(fullTableName, msg.TableNamespace, msg.TableName)
-	return parentTableName != ""
+	return ""
 }
 
 func (c *connector) getParentTableName(fullTableName, tableNamespace, tableName string) string {

elasticsearch/bulk/bulk.go

@@ -9,7 +9,6 @@ import (
 	"time"
 
 	cdc "github.com/Trendyol/go-pq-cdc"
-	"github.com/Trendyol/go-pq-cdc/pq/timescaledb"
 	"github.com/elastic/go-elasticsearch/v7"
 
 	"github.com/Trendyol/go-pq-cdc-elasticsearch/config"
@@ -33,7 +32,7 @@ type Indexer interface {
 		ctx *replication.ListenerContext,
 		eventTime time.Time,
 		actions []elasticsearch2.Action,
-		tableNamespace, tableName string,
+		indexName string,
 		isLastChunk bool,
 	)
 	GetMetric() Metric
@@ -43,7 +42,6 @@ type Indexer interface {
 type Bulk struct {
 	metric              Metric
 	responseHandler     elasticsearch2.ResponseHandler
-	indexMapping        map[string]string
 	config              *config.Config
 	batchKeys           map[string]int
 	batchTicker         *time.Ticker
@@ -91,7 +89,6 @@ func NewBulk(
 		isClosed:            make(chan bool, 1),
 		esClient:            esClient,
 		metric:              NewMetric(pqCDC, config.CDC.Slot.Name),
-		indexMapping:        config.Elasticsearch.TableIndexMapping,
 		config:              config,
 		typeName:            []byte(config.Elasticsearch.TypeName),
 		readers:             readers,
@@ -118,12 +115,12 @@ func (b *Bulk) AddActions(
 	ctx *replication.ListenerContext,
 	eventTime time.Time,
 	actions []elasticsearch2.Action,
-	tableNamespace, tableName string,
+	indexName string,
 	isLastChunk bool,
 ) {
 	b.flushLock.Lock()
 	for i, action := range actions {
-		indexName := b.getIndexName(tableNamespace, tableName, action.IndexName)
+		indexName := b.getIndexName(indexName, action.IndexName)
 		actions[i].IndexName = indexName
 		value := getEsActionJSON(
 			action.ID,
@@ -132,6 +129,7 @@ func (b *Bulk) AddActions(
 			action.Routing,
 			action.Source,
 			b.typeName,
+			b.config.Elasticsearch.Version,
 		)
 
 		b.metric.incrementOp(action.Type, indexName)
@@ -185,7 +183,21 @@ var metaPool = sync.Pool{
 	},
 }
 
-func getEsActionJSON(docID []byte, action elasticsearch2.ActionType, indexName string, routing *string, source []byte, typeName []byte) []byte {
+func isTypeSupported(version string) bool {
+	if version == "" {
+		return true
+	}
+
+	parts := strings.Split(version, ".")
+	if len(parts) == 0 {
+		return true
+	}
+
+	majorVersion := parts[0]
+	return majorVersion < "8"
+}
+
+func getEsActionJSON(docID []byte, action elasticsearch2.ActionType, indexName string, routing *string, source []byte, typeName []byte, esVersion string) []byte {
 	meta := metaPool.Get().([]byte)[:0]
 
 	if action == elasticsearch2.Index {
@@ -200,7 +212,7 @@ func getEsActionJSON(docID []byte, action elasticsearch2.ActionType, indexName s
 		meta = append(meta, routingPrefix...)
 		meta = append(meta, []byte(*routing)...)
 	}
-	if typeName != nil {
+	if typeName != nil && isTypeSupported(esVersion) {
 		meta = append(meta, typePrefix...)
 		meta = append(meta, typeName...)
 	}
@@ -340,25 +352,13 @@ func joinErrors(body map[string]any) (map[string]string, error) {
 	return ivd, fmt.Errorf(sb.String())
 }
 
-func (b *Bulk) getIndexName(tableNamespace, tableName, actionIndexName string) string {
+func (b *Bulk) getIndexName(indexName, actionIndexName string) string {
 	if actionIndexName != "" {
 		return actionIndexName
 	}
 
-	fullTableName := fmt.Sprintf("%s.%s", tableNamespace, tableName)
-
-	indexName := b.indexMapping[fullTableName]
-	if indexName != "" {
-		return indexName
-	}
-
-	t, ok := timescaledb.HyperTables.Load(fullTableName)
-	if ok {
-		indexName = b.indexMapping[t.(string)]
-	}
-
 	if indexName == "" {
-		panic(fmt.Sprintf("there is no index mapping for table: %s.%s on your configuration", tableNamespace, tableName))
+		panic(fmt.Sprintf("there is no index mapping for table: %s on your configuration", indexName))
 	}
 
 	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
+}