[FLINK-38455][elasticsearch][fix] Fix Elasticsearch Missing required property 'BulkRequest.operations' error

Author: Hisoka-X

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/src/main/java/org/apache/flink/cdc/connectors/elasticsearch/v2/Elasticsearch8AsyncWriter.java

@@ -123,11 +123,21 @@ protected void submitRequestEntries(
         LOG.debug("submitRequestEntries with {} items", requestEntries.size());
 
         BulkRequest.Builder br = new BulkRequest.Builder();
+        boolean hasOperations = false;
         for (Operation operation : requestEntries) {
             if (operation.getBulkOperationVariant() == null) {
                 continue;
             }
             br.operations(new BulkOperation(operation.getBulkOperationVariant()));
+            hasOperations = true;
+        }
+
+        if (!hasOperations) {
+            LOG.debug(
+                    "Skipping empty BulkRequest, all {} operation(s) have null BulkOperationVariant",
+                    requestEntries.size());
+            requestResult.accept(Collections.emptyList());
+            return;
         }
 
         esClient.bulk(br.build())

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/src/test/java/org/apache/flink/cdc/connectors/elasticsearch/sink/ElasticsearchDataSinkITCaseTest.java

@@ -146,6 +146,14 @@ void testElasticsearchAddColumn() throws Exception {
         verifyInsertedDataWithNewColumn(tableId, "3", 3, 3.0, "value3", true);
     }
 
+    @Test
+    void testElasticsearchSinkWithOnlySchemaChangeEvents() throws Exception {
+        TableId tableId = TableId.tableId("default", "schema", "schema_only_table");
+        List<Event> events = ElasticsearchTestUtils.createTestEventsWithOnlySchemaChange(tableId);
+
+        runJobWithEvents(events);
+    }
+
     private static ElasticsearchContainer createElasticsearchContainer() {
         ElasticsearchContainer esContainer = new ElasticsearchContainer(ELASTICSEARCH_VERSION);
         esContainer.withLogConsumer(new Slf4jLogConsumer(LOG));

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/src/test/java/org/apache/flink/cdc/connectors/elasticsearch/sink/ElasticsearchEventSerializerTest.java

@@ -90,6 +90,21 @@ void testTableShardingWithSeparator() {
         assertThat(index).isEqualTo("test$2025-01-01");
     }
 
+    @Test
+    void testSchemaChangeEventReturnsNull() {
+        Schema tableSchema =
+                Schema.newBuilder()
+                        .physicalColumn("id", DataTypes.INT().notNull())
+                        .physicalColumn("name", DataTypes.VARCHAR(255).notNull())
+                        .primaryKey("id")
+                        .build();
+
+        ElasticsearchEventSerializer serializer =
+                new ElasticsearchEventSerializer(ZoneId.of("UTC"));
+        CreateTableEvent createTableEvent = new CreateTableEvent(tableId, tableSchema);
+        assertThat(serializer.apply(createTableEvent, new MockContext())).isNull();
+    }
+
     private String getShardingString(Map<TableId, String> shardingKey, String shardingSeparator) {
         RowType rowType =
                 RowType.of(

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-elasticsearch/src/test/java/org/apache/flink/cdc/connectors/elasticsearch/sink/utils/ElasticsearchTestUtils.java

@@ -186,4 +186,24 @@ public static List<Event> createTestEventsWithAddColumn(TableId tableId) {
                                     3, 3.0, BinaryStringData.fromString("value3"), true
                                 })));
     }
+
+    /**
+     * Creates a list of test events that only contain SchemaChangeEvents (CreateTableEvent) without
+     * any DataChangeEvents. This simulates the scenario where a batch only has schema change
+     * events, which previously caused "Missing required property 'BulkRequest.operations'" error.
+     *
+     * @param tableId the identifier of the table.
+     * @return a list of events containing only CreateTableEvent.
+     */
+    public static List<Event> createTestEventsWithOnlySchemaChange(TableId tableId) {
+        Schema schema =
+                Schema.newBuilder()
+                        .column(new PhysicalColumn("id", DataTypes.INT().notNull(), null))
+                        .column(new PhysicalColumn("number", DataTypes.DOUBLE(), null))
+                        .column(new PhysicalColumn("name", DataTypes.VARCHAR(17), null))
+                        .primaryKey("id")
+                        .build();
+
+        return Collections.singletonList(new CreateTableEvent(tableId, schema));
+    }
 }