[Bugfix][Elasticsearch] Fix add column event (#9069)

Author: hailin0

seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/sink/ElasticsearchSinkWriter.java

@@ -23,11 +23,14 @@
 import org.apache.seatunnel.api.sink.SupportSchemaEvolutionSinkWriter;
 import org.apache.seatunnel.api.table.catalog.CatalogTable;
 import org.apache.seatunnel.api.table.catalog.Column;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
 import org.apache.seatunnel.api.table.converter.BasicTypeDefine;
 import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
 import org.apache.seatunnel.api.table.schema.event.AlterTableColumnEvent;
 import org.apache.seatunnel.api.table.schema.event.AlterTableColumnsEvent;
 import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
+import org.apache.seatunnel.api.table.schema.handler.TableSchemaChangeEventDispatcher;
+import org.apache.seatunnel.api.table.schema.handler.TableSchemaChangeEventHandler;
 import org.apache.seatunnel.api.table.type.RowKind;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated;
@@ -66,12 +69,14 @@ public class ElasticsearchSinkWriter
 
     private final int maxBatchSize;
 
-    private final SeaTunnelRowSerializer seaTunnelRowSerializer;
+    private SeaTunnelRowSerializer seaTunnelRowSerializer;
     private final List<String> requestEsList;
     private EsRestClient esRestClient;
     private RetryMaterial retryMaterial;
     private static final long DEFAULT_SLEEP_TIME_MS = 200L;
     private final IndexInfo indexInfo;
+    private TableSchema tableSchema;
+    private final TableSchemaChangeEventHandler tableSchemaChangeEventHandler;
 
     public ElasticsearchSinkWriter(
             Context context,
@@ -94,6 +99,8 @@ public ElasticsearchSinkWriter(
         this.requestEsList = new ArrayList<>(maxBatchSize);
         this.retryMaterial =
                 new RetryMaterial(maxRetryCount, true, exception -> true, DEFAULT_SLEEP_TIME_MS);
+        this.tableSchema = catalogTable.getTableSchema();
+        this.tableSchemaChangeEventHandler = new TableSchemaChangeEventDispatcher();
     }
 
     @Override
@@ -120,6 +127,13 @@ public void applySchemaChange(SchemaChangeEvent event) throws IOException {
         } else {
             throw new UnsupportedOperationException("Unsupported alter table event: " + event);
         }
+
+        this.tableSchema = tableSchemaChangeEventHandler.reset(tableSchema).apply(event);
+        this.seaTunnelRowSerializer =
+                new ElasticsearchRowSerializer(
+                        esRestClient.getClusterInfo(),
+                        indexInfo,
+                        tableSchema.toPhysicalRowDataType());
     }
 
     private void applySingleSchemaChangeEvent(SchemaChangeEvent event) {

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-elasticsearch-e2e/src/test/java/org/apache/seatunnel/e2e/connector/elasticsearch/ElasticsearchSchemaChangeIT.java

@@ -192,9 +192,30 @@ public void testSchemaChange(TestContainer container) throws InterruptedExceptio
                                     this.container.execInContainer(
                                             "bash",
                                             "-c",
-                                            "curl -k -u elastic:elasticsearch https://localhost:9200/schema_change_index/_count");
-                            Assertions.assertTrue(
-                                    indexCountResult.getStdout().contains("\"count\":18"));
+                                            "curl -k -u elastic:elasticsearch -H \"Content-Type:application/json\" -d '{ \"from\": 0, \"size\": 10000, \"query\": { \"match_all\": {}}}' https://localhost:9200/schema_change_index/_search");
+                            log.info("indexCountResult: {}", indexCountResult.getStdout());
+                            ObjectNode jsonNode =
+                                    JsonUtils.parseObject(indexCountResult.getStdout());
+                            JsonNode hits = jsonNode.get("hits");
+                            long totalCount = hits.get("total").get("value").asLong();
+                            Assertions.assertEquals(18L, totalCount);
+
+                            hits.get("hits")
+                                    .forEach(
+                                            hit -> {
+                                                JsonNode source = hit.get("_source");
+                                                int id = source.get("id").asInt();
+                                                if (id >= 119 && id <= 127) {
+                                                    Assertions.assertTrue(
+                                                            source.has("add_column1"));
+                                                    Assertions.assertFalse(
+                                                            source.get("add_column1").isNull());
+                                                    Assertions.assertTrue(
+                                                            source.has("add_column2"));
+                                                    Assertions.assertFalse(
+                                                            source.get("add_column2").isNull());
+                                                }
+                                            });
                         });
     }