Refactor gRPC converters to use registry pattern

This commit implements an extensible registry pattern for both response-side
and request-side gRPC converters, improving consistency and maintainability.

Response-side changes:
- Add AggregateProtoConverter and AggregateProtoConverterRegistry SPI interfaces
- Implement AggregateProtoConverterSpiRegistry with class hierarchy lookup
- Create AggregateProtoConverterRegistryImpl as public facade
- Add MinAggregateProtoConverter and MaxAggregateProtoConverter implementations
- Update AggregateProtoUtils to delegate to registry instead of instanceof checks
- Add comprehensive test suite (AggregateProtoConverterRegistryTests)

Request-side changes:
- Inline ObjectParserProtoUtils logic into ValuesSourceAggregationProtoUtils
- Remove unnecessary ObjectParserProtoUtils abstraction layer
- Simplify field parsing by directly calling builder methods

Benefits:
- Extensible: New aggregation types can be added without modifying central dispatcher
- Consistent: Response-side now mirrors request-side registry pattern
- Type-safe: Class-based dispatch with support for subclasses and mocks
- Simplified: Reduced complexity by removing one abstraction layer

All 1,280+ transport-grpc tests passing.

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

Author: yiyuabc

modules/transport-grpc/spi/src/main/java/org/opensearch/transport/grpc/spi/AggregateProtoConverter.java

@@ -0,0 +1,42 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.transport.grpc.spi;
+
+import org.opensearch.protobufs.Aggregate;
+import org.opensearch.search.aggregations.InternalAggregation;
+
+import java.io.IOException;
+
+/**
+ * SPI interface for converting OpenSearch InternalAggregation objects to Protocol Buffer Aggregate messages.
+ * Follows the same pattern as {@link AggregationBuilderProtoConverter} for request-side conversions.
+ *
+ * <p>The registry handles metadata centrally. Converters should delegate to existing
+ * {@code *AggregateProtoUtils} classes for the actual conversion logic.
+ */
+public interface AggregateProtoConverter {
+
+    /**
+     * Returns the InternalAggregation subclass this converter handles.
+     *
+     * @return The class type of the aggregation this converter supports
+     */
+    Class<? extends InternalAggregation> getHandledAggregationType();
+
+    /**
+     * Converts an InternalAggregation to its Protocol Buffer Aggregate.Builder representation.
+     * Returns a builder to allow the registry to construct the final Aggregate.
+     * Mirrors REST-side {@link org.opensearch.search.aggregations.InternalAggregation#toXContent}
+     *
+     * @param aggregation The InternalAggregation to convert (guaranteed to be of the handled type)
+     * @return An Aggregate.Builder with the aggregation-specific fields populated
+     * @throws IOException if an error occurs during protobuf conversion
+     */
+    Aggregate.Builder toProto(InternalAggregation aggregation) throws IOException;
+}

modules/transport-grpc/spi/src/main/java/org/opensearch/transport/grpc/spi/AggregateProtoConverterRegistry.java

@@ -0,0 +1,32 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.transport.grpc.spi;
+
+import org.opensearch.protobufs.Aggregate;
+import org.opensearch.search.aggregations.InternalAggregation;
+
+import java.io.IOException;
+
+/**
+ * SPI interface for the aggregate converter registry.
+ * Provides the main entry point for converting InternalAggregation objects to Protocol Buffer Aggregate messages.
+ */
+public interface AggregateProtoConverterRegistry {
+
+    /**
+     * Converts an InternalAggregation to its Protocol Buffer Aggregate representation.
+     * Handles metadata and delegates to the appropriate converter.
+     *
+     * @param aggregation The InternalAggregation to convert (must not be null)
+     * @return The corresponding Protocol Buffer Aggregate message
+     * @throws IllegalArgumentException if aggregation is null or type is not supported
+     * @throws IOException if an error occurs during protobuf conversion
+     */
+    Aggregate toProto(InternalAggregation aggregation) throws IOException;
+}

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/request/common/ObjectParserProtoUtils.java

@@ -11,11 +11,8 @@
 import org.opensearch.script.Script;
 import org.opensearch.search.aggregations.support.ValuesSourceAggregationBuilder;
 import org.opensearch.search.aggregations.support.ValueType;
-import org.opensearch.transport.grpc.proto.request.common.ObjectParserProtoUtils;
 import org.opensearch.transport.grpc.util.ProtobufEnumUtils;
 
-import java.util.function.Function;
-
 /**
  * Utility class for parsing common fields from ValuesSource-based aggregation Protocol Buffer messages.
  */
@@ -28,7 +25,7 @@ private ValuesSourceAggregationProtoUtils() {
     /**
      * Declares common fields for ValuesSource aggregations, mirroring {@link ValuesSourceAggregationBuilder#declareFields}.
      *
-     * <p>Uses {@link ObjectParserProtoUtils} for explicit, declarative field parsing that mirrors REST's ObjectParser pattern.
+     * <p>Directly sets fields on the builder, mirroring REST's ObjectParser pattern.
      *
      * @param builder The aggregation builder to set fields on
      * @param fields Wrapper containing all ValuesSource field values from proto
@@ -50,7 +47,7 @@ public static void declareFields(
     /**
      * Declares common fields for ValuesSource aggregations, mirroring {@link ValuesSourceAggregationBuilder#declareFields}.
      *
-     * <p>Uses {@link ObjectParserProtoUtils} for explicit, declarative field parsing that mirrors REST's ObjectParser pattern.
+     * <p>Directly sets field values when present, mirroring REST's ObjectParser behavior without the abstraction layer.
      *
      * @param builder The aggregation builder to set fields on
      * @param fields Wrapper containing all ValuesSource field values from proto
@@ -68,60 +65,65 @@ public static void declareFields(
         boolean timezoneAware,
         boolean fieldRequired
     ) {
-        ObjectParserProtoUtils.declareField(
-            builder,
-            ValuesSourceAggregationBuilder::field,
-            fields.getField(),
-            Function.identity(),
-            ParseField.CommonFields.FIELD.getPreferredName()
-        );
+        // Field declaration - mirrors REST ObjectParser.declareField behavior
+        if (fields.getField() != null) {
+            try {
+                builder.field(fields.getField());
+            } catch (Exception e) {
+                throw new IllegalArgumentException("Failed to parse [field]", e);
+            }
+        }
 
-        ObjectParserProtoUtils.declareField(
-            builder,
-            ValuesSourceAggregationBuilder::missing,
-            fields.getMissing(),
-            Function.identity(),
-            ParseField.CommonFields.MISSING.getPreferredName()
-        );
+        // Missing value declaration
+        if (fields.getMissing() != null) {
+            try {
+                builder.missing(fields.getMissing());
+            } catch (Exception e) {
+                throw new IllegalArgumentException("Failed to parse [missing]", e);
+            }
+        }
 
-        ObjectParserProtoUtils.declareField(
-            builder,
-            ValuesSourceAggregationBuilder::userValueTypeHint,
-            fields.getValueTypeProto(),
-            proto -> {
-                String valueTypeStr = ProtobufEnumUtils.convertToString(proto);
-                if (valueTypeStr == null) {
-                    return null;
-                }
-                ValueType valueType = ValueType.lenientParse(valueTypeStr);
-                if (valueType == null) {
-                    throw new IllegalArgumentException("Unknown value type [" + valueTypeStr + "]");
+        // Value type declaration with transformation
+        if (fields.getValueTypeProto() != null) {
+            try {
+                String valueTypeStr = ProtobufEnumUtils.convertToString(fields.getValueTypeProto());
+                if (valueTypeStr != null) {
+                    ValueType valueType = ValueType.lenientParse(valueTypeStr);
+                    if (valueType == null) {
+                        throw new IllegalArgumentException("Unknown value type [" + valueTypeStr + "]");
+                    }
+                    builder.userValueTypeHint(valueType);
                 }
-                return valueType;
-            },
-            ValueType.VALUE_TYPE.getPreferredName()
-        );
+            } catch (IllegalArgumentException e) {
+                throw e;
+            } catch (Exception e) {
+                throw new IllegalArgumentException("Failed to parse [value_type]", e);
+            }
+        }
 
+        // Format declaration (conditional)
         if (formattable) {
-            ObjectParserProtoUtils.declareField(
-                builder,
-                ValuesSourceAggregationBuilder::format,
-                fields.getFormat(),
-                Function.identity(),
-                ParseField.CommonFields.FORMAT.getPreferredName()
-            );
+            if (fields.getFormat() != null) {
+                try {
+                    builder.format(fields.getFormat());
+                } catch (Exception e) {
+                    throw new IllegalArgumentException("Failed to parse [format]", e);
+                }
+            }
         }
 
+        // Field/script requirement validation
         if (scriptable) {
-            ObjectParserProtoUtils.declareField(
-                builder,
-                ValuesSourceAggregationBuilder::script,
-                fields.getScript(),
-                Function.identity(),
-                Script.SCRIPT_PARSE_FIELD.getPreferredName()
-            );
+            if (fields.getScript() != null) {
+                try {
+                    builder.script(fields.getScript());
+                } catch (Exception e) {
+                    throw new IllegalArgumentException("Failed to parse [script]", e);
+                }
+            }
 
             if (fieldRequired) {
+                // throw exception when neither field nor script is specified
                 if (fields.getField() == null && fields.getScript() == null) {
                     throw new IllegalArgumentException(
                         "Required one of fields ["

modules/transport-grpc/src/main/java/org/opensearch/transport/grpc/proto/request/search/aggregation/support/ValuesSourceAggregationProtoUtils.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");