[Feature][Integration][Java] Add Milvus vector store integration

Signed-off-by: YangYanbin <warlock.yyb@alibaba-inc.com>

Author: yanbinyang

.github/workflows/ci.yml

@@ -265,9 +265,18 @@ jobs:
         run: bash tools/build.sh
       - name: Install ollama
         run: bash tools/start_ollama_server.sh
+      - name: Start Milvus
+        run: |
+          docker compose -f tools/docker/milvus/docker-compose.yml down -v
+          docker compose -f tools/docker/milvus/docker-compose.yml up -d
+          timeout 180 bash -c 'until curl -fsS http://localhost:9091/healthz; do sleep 5; done'
       - name: Run e2e tests
         env:
           LOG_LEVEL: INFO
         run: |
           export ES_HOST="http://localhost:9200"
-          tools/e2e.sh
+          export MILVUS_URI="http://localhost:19530"
+          tools/e2e.sh
+      - name: Stop Milvus
+        if: always()
+        run: docker compose -f tools/docker/milvus/docker-compose.yml down -v

api/src/main/java/org/apache/flink/agents/api/resource/ResourceName.java

@@ -171,6 +171,10 @@ public static final class VectorStore {
         public static final String ELASTICSEARCH_VECTOR_STORE =
                 "org.apache.flink.agents.integrations.vectorstores.elasticsearch.ElasticsearchVectorStore";
 
+        // Milvus
+        public static final String MILVUS_VECTOR_STORE =
+                "org.apache.flink.agents.integrations.vectorstores.milvus.MilvusVectorStore";
+
         // Python Wrapper
         public static final String PYTHON_WRAPPER_VECTOR_STORE =
                 "org.apache.flink.agents.api.vectorstores.python.PythonVectorStore";

dist/pom.xml

@@ -100,6 +100,11 @@ under the License.
             <artifactId>flink-agents-integrations-vector-stores-elasticsearch</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-agents-integrations-vector-stores-milvus</artifactId>
+            <version>${project.version}</version>
+        </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-agents-integrations-vector-stores-opensearch</artifactId>
@@ -156,4 +161,4 @@ under the License.
             </plugin>
         </plugins>
     </build>
-</project>
+</project>

docs/content/docs/development/vector_stores.md

@@ -41,6 +41,7 @@ In Flink Agents, vector stores are essential for:
 
 ### Concepts
 * **Document**: Document is the abstraction that represents a piece of text and associated metadata. A document may also carry a pre-computed `embedding` vector and a `score` populated by query results.
+* **Collection**: Collection is the abstraction that represents a set of documents. It corresponds to different concept for different vector store specification, like index in Elasticsearch/OpenSearch and collection in Chroma/Milvus.
 * **Filter DSL**: A unified, equality-only metadata filter dialect shared by `query`, `get`, and `delete`. The DSL covers only the subset every supported backend can honour (equality matching), so callers don't need to know each store's native operators. See the [Filter DSL](#filter-dsl) section below for details.
 
 ## How to use
@@ -172,7 +173,7 @@ For vector stores that implement `CollectionManageableVectorStore`, you can crea
 * `delete_collection` / `deleteCollection`: Delete a collection by name.
 
 {{< hint info >}}
-Collection-level operations are only supported for vector stores that implement `CollectionManageableVectorStore`. Among the built-in providers, Chroma (Python), Elasticsearch (Java) and OpenSearch (Java) implement this interface.
+Collection-level operations are only supported for vector stores that implement `CollectionManageableVectorStore`. Among the built-in providers, Chroma (Python), Elasticsearch (Java), OpenSearch (Java), and Milvus (Java) implement this interface.
 {{< /hint >}}
 
 {{< tabs "Collection level operations" >}}
@@ -642,9 +643,86 @@ public static ResourceDescriptor vectorStore() {
 
 {{< /tabs >}}
 
+### Milvus
+
+[Milvus](https://milvus.io/) is an open-source vector database designed for high-dimensional vector search at scale.
+
+{{< hint info >}}
+Milvus is currently supported in the Java API only. To use Milvus from Python agents, see [Using Cross-Language Providers](#using-cross-language-providers).
+{{< /hint >}}
+
+#### Prerequisites
+
+1. A Milvus server.
+
+#### MilvusVectorStore Parameters
+
+| Parameter                   | Type | Default                              | Description                                                                 |
+|-----------------------------|------|--------------------------------------|-----------------------------------------------------------------------------|
+| `embedding_model`           | str  | Required                             | Reference to embedding model resource name                                  |
+| `collection`                | str  | `"flink_agents_milvus_collection"`   | Default target Milvus collection name                                       |
+| `collection_name`           | str  | None                                 | Alias for `collection`                                                       |
+| `index`                     | str  | None                                 | Alias for `collection`, mainly for cross-provider compatibility              |
+| `id_field`                  | str  | `"id"`                               | Name of the primary key field                                                |
+| `content_field`             | str  | `"content"`                          | Name of the field storing document content                                   |
+| `metadata_field`            | str  | `"metadata"`                         | Name of the JSON field storing document metadata                             |
+| `vector_field`              | str  | `"embedding"`                        | Name of the FloatVector field used for vector search                         |
+| `dims`                      | int  | `768`                                | Vector dimensionality                                                        |
+| `id_max_length`             | int  | `65535`                              | Maximum length for the VarChar primary key field                             |
+| `content_max_length`        | int  | `65535`                              | Maximum length for the VarChar content field                                 |
+| `metric_type`               | str  | `"COSINE"`                           | Milvus metric type used by vector search                                     |
+| `index_type`                | str  | `"AUTOINDEX"`                        | Milvus vector index type                                                     |
+| `index_params`              | map  | `{}`                                 | Extra vector index parameters passed to Milvus                               |
+| `metadata_index_keys`       | list | `user_id`, `agent_id`, `run_id`, `actor_id`, `category` | Additional metadata JSON keys indexed with path indexes |
+| `metadata_index_cast_types` | map  | Default keys use `"VARCHAR"`         | Per-metadata-key JSON path index cast type overrides                         |
+| `num_shards`                | int  | `1`                                  | Number of Milvus shards for newly created collections                        |
+| `consistency_level`         | str  | `"BOUNDED"`                          | Milvus consistency level for collection creation, query, and search          |
+| `max_get_limit`             | int  | `10000`                              | Maximum number of documents returned by `get` when no limit is specified     |
+| `load_timeout_ms`           | long | `120000`                             | Timeout for loading collections                                              |
+| `uri`                       | str  | `"http://localhost:19530"`           | Milvus endpoint                                                              |
+| `host`                      | str  | `"localhost"`                        | Milvus host used when `uri` is not set                                       |
+| `port`                      | int  | `19530`                              | Milvus port used when `uri` is not set                                       |
+| `db_name`                   | str  | None                                 | Milvus database name                                                         |
+| `token`                     | str  | None                                 | Token for Milvus authentication                                              |
+| `username`                  | str  | None                                 | Username for basic authentication                                            |
+| `password`                  | str  | None                                 | Password for basic authentication                                            |
+| `enable_precheck`           | bool | `false`                              | Whether to enable Milvus client precheck                                     |
+
+{{< hint info >}}
+When creating a collection, MilvusVectorStore creates a primary-key field, content field, JSON metadata field, vector field, vector index, and JSON metadata indexes. The default metadata JSON path indexes cover common filter keys such as `user_id`, `agent_id`, `run_id`, `actor_id`, and `category`; add `metadata_index_keys` for application-specific filter keys.
+
+The default shard count is `1`. As a rough capacity-planning rule, use about one shard per 100 million vectors, and increase it for heavier write throughput.
+{{< /hint >}}
+
+#### Usage Example
+
+{{< tabs "Milvus Usage Example" >}}
+
+{{< tab "Java" >}}
+
+```java
+@VectorStore
+public static ResourceDescriptor vectorStore() {
+    return ResourceDescriptor.Builder.newBuilder(ResourceName.VectorStore.MILVUS_VECTOR_STORE)
+            .addInitialArgument("embedding_model", "embeddingModel")
+            .addInitialArgument("uri", "http://localhost:19530")
+            .addInitialArgument("collection", "my_documents")
+            .addInitialArgument("dims", 1536)
+            .addInitialArgument("metric_type", "COSINE")
+            .addInitialArgument("index_type", "AUTOINDEX")
+            // Optional metadata JSON path indexes
+            // .addInitialArgument("metadata_index_keys", List.of("user_id", "agent_id", "run_id"))
+            .build();
+}
+```
+
+{{< /tab >}}
+
+{{< /tabs >}}
+
 ## Using Cross-Language Providers
 
-Flink Agents supports cross-language vector store integration, allowing you to use vector stores implemented in one language (Java or Python) from agents written in the other language. This is particularly useful when a vector store provider is only available in one language (e.g., Elasticsearch is currently Java-only, Chroma is currently Python-only).
+Flink Agents supports cross-language vector store integration, allowing you to use vector stores implemented in one language (Java or Python) from agents written in the other language. This is particularly useful when a vector store provider is only available in one language (e.g., Elasticsearch and Milvus are currently Java-only, Chroma is currently Python-only).
 
 {{< hint warning >}}
 **Limitations:**
@@ -1101,4 +1179,4 @@ public class MyVectorStore extends BaseVectorStore
 
 {{< /tab >}}
 
-{{< /tabs >}}
+{{< /tabs >}}

docs/content/docs/faq/faq.md

@@ -117,6 +117,7 @@ Flink Agents provides built-in integrations for many ecosystem providers. Some i
 |---|---|---|
 | [Chroma]({{< ref "docs/development/vector_stores#chroma" >}}) | ✅ | ❌ |
 | [Elasticsearch]({{< ref "docs/development/vector_stores#elasticsearch" >}}) | ❌ | ✅ |
+| [Milvus]({{< ref "docs/development/vector_stores#milvus" >}}) | ❌ | ✅ |
 
 **MCP Server**
 
@@ -131,4 +132,4 @@ Flink Agents provides built-in integrations for many ecosystem providers. Some i
 To avoid potential conflict with Flink cluster, the scope of the dependencies related to Flink and Flink Agents for agent job are provided. See [Maven Dependencies]({{< ref "docs/get-started/installation#maven-dependencies-for-java" >}}) for details.
 
 To run the examples in IDE, users must enable the IDE feature: `add dependencies with provided scope to classpath`.
-* For **IDEA**, edit the **`Run/Debug Configuration`** and enable **`add dependencies with provided scope to classpath`**. See [Run/Debug Configuration](https://www.jetbrains.com/help/idea/run-debug-configuration-scala.html) for details.
+* For **IDEA**, edit the **`Run/Debug Configuration`** and enable **`add dependencies with provided scope to classpath`**. See [Run/Debug Configuration](https://www.jetbrains.com/help/idea/run-debug-configuration-scala.html) for details.

e2e-test/flink-agents-end-to-end-tests-resource-cross-language/src/test/java/org/apache/flink/agents/resource/test/VectorStoreCrossLanguageAgent.java

@@ -59,6 +59,7 @@
 public class VectorStoreCrossLanguageAgent extends Agent {
     public static final String OLLAMA_MODEL = "nomic-embed-text";
     public static final String TEST_COLLECTION = "test_collection";
+    private static final String VECTOR_STORE_BACKEND = "CHROMA";
 
     @EmbeddingModelConnection
     public static ResourceDescriptor embeddingConnection() {
@@ -121,7 +122,8 @@ public static void inputEvent(Event event, RunnerContext ctx) throws Exception {
                     TEST_COLLECTION,
                     Map.of("metadata", Map.of("key1", "value1", "key2", "value2")));
 
-            System.out.println("[TEST] Vector store Collection Management PASSED");
+            System.out.printf(
+                    "[TEST][%s] Vector store Collection Management PASSED%n", VECTOR_STORE_BACKEND);
 
             vectorStore.deleteCollection(TEST_COLLECTION);
             Assertions.assertThrows(
@@ -168,7 +170,8 @@ public static void inputEvent(Event event, RunnerContext ctx) throws Exception {
             Assertions.assertEquals(
                     Map.of("category", "database", "source", "test"), doc.getMetadata());
 
-            System.out.println("[TEST] Vector store Document Management PASSED");
+            System.out.printf(
+                    "[TEST][%s] Vector store Document Management PASSED%n", VECTOR_STORE_BACKEND);
 
             // Verify VectorStoreQuery.filters survives the Java->Python bridge.
             // ChromaDB applies the unified-DSL filter to its `where` clause, so the
@@ -191,7 +194,8 @@ public static void inputEvent(Event event, RunnerContext ctx) throws Exception {
                     filteredDocs.get(0).getId(),
                     "Filter {category=database} should match doc2");
 
-            System.out.println("[TEST] Vector store filter query PASSED");
+            System.out.printf(
+                    "[TEST][%s] Vector store filter query PASSED%n", VECTOR_STORE_BACKEND);
 
             ctx.getShortTermMemory().set("is_initialized", true);
         }
@@ -244,12 +248,16 @@ public static void contextRetrievalResponseEvent(Event event, RunnerContext ctx)
                     first.getContent().substring(0, Math.min(50, first.getContent().length())));
 
             ctx.sendEvent(new OutputEvent(result));
-            System.out.printf("[TEST] Vector store retrieval PASSED, count=%d%n", documents.size());
+            System.out.printf(
+                    "[TEST][%s] Vector store retrieval PASSED, count=%d%n",
+                    VECTOR_STORE_BACKEND, documents.size());
         } catch (Exception e) {
             result.put("test_status", "FAILED");
             result.put("error", e.getMessage());
             ctx.sendEvent(new OutputEvent(result));
-            System.err.printf("[TEST] Vector store retrieval FAILED: %s%n", e.getMessage());
+            System.err.printf(
+                    "[TEST][%s] Vector store retrieval FAILED: %s%n",
+                    VECTOR_STORE_BACKEND, e.getMessage());
             throw e;
         }
     }

integrations/pom.xml

@@ -33,6 +33,7 @@ under the License.
     <properties>
         <ollama4j.version>1.1.5</ollama4j.version>
         <elasticsearch.version>8.19.0</elasticsearch.version>
+        <milvus.version>2.6.18</milvus.version>
         <openai.version>4.8.0</openai.version>
         <anthropic.version>2.11.1</anthropic.version>
         <aws.sdk.version>2.32.16</aws.sdk.version>

integrations/vector-stores/milvus/pom.xml

@@ -0,0 +1,52 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you 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
+  http://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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>org.apache.flink</groupId>
+        <artifactId>flink-agents-integrations-vector-stores</artifactId>
+        <version>0.3-SNAPSHOT</version>
+        <relativePath>../pom.xml</relativePath>
+    </parent>
+
+    <artifactId>flink-agents-integrations-vector-stores-milvus</artifactId>
+    <name>Flink Agents : Integrations: Vector Stores: Milvus</name>
+    <packaging>jar</packaging>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.flink</groupId>
+            <artifactId>flink-agents-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.milvus</groupId>
+            <artifactId>milvus-sdk-java</artifactId>
+            <version>${milvus.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+            <version>${slf4j.version}</version>
+        </dependency>
+    </dependencies>
+
+</project>