Add plugin support for response-side aggregate converters

Enable plugins to register custom AggregateProtoConverter implementations
for serializing custom aggregation results to protobuf responses.

Changes:
- Load AggregateProtoConverter extensions from plugins in loadExtensions()
- Create and initialize AggregateProtoConverterRegistryImpl in createComponents()
- Set global registry via AggregateProtoUtils.setRegistry()
- Register external converters in the SPI registry
- Add logging for discovery and registration of aggregate converters

This completes the extensibility story - plugins can now:
1. Register custom request converters (AggregationBuilderProtoConverter)
2. Register custom response converters (AggregateProtoConverter)

Both directions are now fully pluggable and extensible.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: yiyuabc

modules/transport-grpc/src/main/java/org/opensearch/transport/grpc/GrpcPlugin.java

@@ -36,8 +36,11 @@
 import org.opensearch.transport.grpc.proto.request.search.aggregation.AggregationBuilderProtoConverterRegistryImpl;
 import org.opensearch.transport.grpc.proto.request.search.query.AbstractQueryBuilderProtoUtils;
 import org.opensearch.transport.grpc.proto.request.search.query.QueryBuilderProtoConverterRegistryImpl;
+import org.opensearch.transport.grpc.proto.response.search.aggregation.AggregateProtoConverterRegistryImpl;
+import org.opensearch.transport.grpc.proto.response.search.aggregation.AggregateProtoUtils;
 import org.opensearch.transport.grpc.services.DocumentServiceImpl;
 import org.opensearch.transport.grpc.services.SearchServiceImpl;
+import org.opensearch.transport.grpc.spi.AggregateProtoConverter;
 import org.opensearch.transport.grpc.spi.AggregationBuilderProtoConverter;
 import org.opensearch.transport.grpc.spi.GrpcInterceptorProvider;
 import org.opensearch.transport.grpc.spi.GrpcInterceptorProvider.OrderedGrpcInterceptor;
@@ -85,9 +88,11 @@ public final class GrpcPlugin extends Plugin implements NetworkPlugin, Extensibl
 
     private final List<QueryBuilderProtoConverter> queryConverters = new ArrayList<>();
     private final List<AggregationBuilderProtoConverter> aggregationConverters = new ArrayList<>();
+    private final List<AggregateProtoConverter> aggregateConverters = new ArrayList<>();
     private final List<GrpcServiceFactory> servicesFactory = new ArrayList<>();
     private QueryBuilderProtoConverterRegistryImpl queryRegistry;
     private AggregationBuilderProtoConverterRegistryImpl aggregationRegistry;
+    private AggregateProtoConverterRegistryImpl aggregateRegistry;
     private AbstractQueryBuilderProtoUtils queryUtils;
     private GrpcInterceptorChain serverInterceptor; // Initialized in createComponents
     private List<GrpcInterceptorProvider> interceptorProviders = new ArrayList<>();
@@ -140,6 +145,23 @@ public void loadExtensions(ExtensiblePlugin.ExtensionLoader loader) {
             logger.info("No AggregationBuilderProtoConverter extensions found from other plugins");
         }
 
+        // Load aggregate converters (response-side) from other plugins
+        List<AggregateProtoConverter> aggResponseExtensions = loader.loadExtensions(AggregateProtoConverter.class);
+        if (aggResponseExtensions != null && !aggResponseExtensions.isEmpty()) {
+            logger.info("Loading {} AggregateProtoConverter extensions from other plugins", aggResponseExtensions.size());
+            for (AggregateProtoConverter converter : aggResponseExtensions) {
+                logger.info(
+                    "Discovered AggregateProtoConverter extension: {} (handles: {})",
+                    converter.getClass().getName(),
+                    converter.getHandledAggregationType().getName()
+                );
+                aggregateConverters.add(converter);
+            }
+            logger.info("Successfully loaded {} AggregateProtoConverter extensions", aggResponseExtensions.size());
+        } else {
+            logger.info("No AggregateProtoConverter extensions found from other plugins");
+        }
+
         List<GrpcInterceptorProvider> providers = loader.loadExtensions(GrpcInterceptorProvider.class);
         if (providers != null) {
             // Note: ThreadContext will be provided during component creation
@@ -417,6 +439,12 @@ public Collection<Object> createComponents(
         // Create the aggregation registry
         this.aggregationRegistry = new AggregationBuilderProtoConverterRegistryImpl();
 
+        // Create the aggregate registry (response-side)
+        this.aggregateRegistry = new AggregateProtoConverterRegistryImpl();
+
+        // Set the global aggregate registry for response serialization
+        AggregateProtoUtils.setRegistry(aggregateRegistry);
+
         // Create the query utils instance
         this.queryUtils = new AbstractQueryBuilderProtoUtils(queryRegistry);
 
@@ -475,6 +503,28 @@ public Collection<Object> createComponents(
             logger.info("No external AggregationBuilderProtoConverter(s) to register");
         }
 
+        // Register external aggregate converters (response-side)
+        if (!aggregateConverters.isEmpty()) {
+            logger.info(
+                "Registering {} external AggregateProtoConverter(s)",
+                aggregateConverters.size()
+            );
+            for (AggregateProtoConverter converter : aggregateConverters) {
+                logger.info(
+                    "Processing external aggregate converter: {} (handles: {})",
+                    converter.getClass().getName(),
+                    converter.getHandledAggregationType().getName()
+                );
+
+                // Register the converter in the SPI registry
+                aggregateRegistry.getSpiRegistry().registerConverter(converter);
+                logger.info("Registered aggregate converter: {}", converter.getClass().getName());
+            }
+            logger.info("Successfully registered all {} external aggregate converters", aggregateConverters.size());
+        } else {
+            logger.info("No external AggregateProtoConverter(s) to register");
+        }
+
         return super.createComponents(
             client,
             clusterService,

modules/transport-grpc/src/main/java/org/opensearch/transport/grpc/proto/response/search/SearchResponseSectionsProtoUtils.java

@@ -85,8 +85,6 @@ protected static void toProto(org.opensearch.protobufs.SearchResponse.Builder bu
      * @throws UnsupportedOperationException if unsupported features are present
      */
     private static void checkUnsupportedFeatures(SearchResponse response) {
-        // Aggregations are now supported, removed from unsupported list
-
         // TODO: Implement suggest conversion
         if (response.getSuggest() != null) {
             throw new UnsupportedOperationException("suggest responses are not supported yet");