review comments

Author: alkatrivedi

handwritten/spanner/google-cloud-spanner-executor/src/cloud-client-executor.ts

@@ -84,10 +84,22 @@ export class ExecutionFlowContext implements ExecutionFlowContextInterface {
   }
 }
 
+type ActionHandler = (action: any, sender: OutcomeSender) => Promise<void>;
+
 export class CloudClientExecutor {
   private spanner: Spanner;
   private tracer: Tracer;
 
+  private readonly adminActionRegistry: Record<string, ActionHandler> = {
+    createCloudInstance: (action, sender) =>
+      this.executeCreateCloudInstance(action as ICreateCloudInstanceAction, sender),
+  };
+
+  private readonly actionRegistry: Record<string, ActionHandler> = {
+    admin: (action, sender) =>
+      this.executeAdminAction(action as IAdminAction, sender),
+  };
+
   constructor() {
     const spannerOptions = CloudUtil.getSpannerOptions();
     this.spanner = new Spanner(spannerOptions);
@@ -144,8 +156,9 @@ export class CloudClientExecutor {
 
     return context.with(trace.setSpan(context.active(), span), async () => {
       try {
-        if (action.admin) {
-          await this.executeAdminAction(action.admin, outcomeSender);
+        const handler = this.actionRegistry[actionType];
+        if (handler) {
+          await handler(action[actionType as keyof typeof action], outcomeSender);
           return;
         }
 
@@ -171,16 +184,21 @@ export class CloudClientExecutor {
     sender: OutcomeSender,
   ): Promise<void> {
     try {
-      if (action.createCloudInstance) {
-        await this.executeCreateCloudInstance(
-          action.createCloudInstance,
+      const adminType = Object.keys(action).find(
+        k => action[k as keyof typeof action] !== undefined,
+      );
+
+      if (adminType && this.adminActionRegistry[adminType]) {
+        await this.adminActionRegistry[adminType](
+          action[adminType as keyof typeof action],
           sender,
         );
         return;
       }
+
       sender.finishWithError({
         code: status.UNIMPLEMENTED,
-        message: 'Admin action not implemented',
+        message: `Admin action ${adminType || 'unknown'} not implemented`,
       });
     } catch (e: any) {
       sender.finishWithError(e);
@@ -213,7 +231,6 @@ export class CloudClientExecutor {
       });
 
       console.log('Waiting for instance creation operation to complete...');
-
       await operation.promise();
 
       console.log(`Instance ${instanceId} created successfully.`);

handwritten/spanner/google-cloud-spanner-executor/src/cloud-executor-impl.ts

@@ -14,7 +14,7 @@
  * limitations under the License.
  */
 
-import {ServerDuplexStream, status, ServiceError} from '@grpc/grpc-js';
+import { ServerDuplexStream, status } from '@grpc/grpc-js';
 import {trace, context, Tracer} from '@opentelemetry/api';
 import {CloudClientExecutor} from './cloud-client-executor';
 import * as protos from '../../protos/protos';
@@ -64,22 +64,7 @@ export class CloudExecutorImpl {
     call.on('data', (request: SpannerAsyncActionRequest) => {
       context.with(streamContext, () => {
         console.log(`Receiving request: \n${JSON.stringify(request, null, 2)}`);
-
-        // Ensure nested properties exist before attempting to inject configuration overrides.
-        if (!request.action) request.action = {};
-        if (!request.action.spannerOptions) request.action.spannerOptions = {};
-        if (!request.action.spannerOptions.sessionPoolOptions)
-          request.action.spannerOptions.sessionPoolOptions = {};
-
-        // as the multiplexed session is defualt enabled, mutate the incoming request to forcefully enforce multiplexed sessions for this proxy execution.
-        request.action.spannerOptions.sessionPoolOptions.useMultiplexed = true;
-
-        console.log(
-          `Updated request to set multiplexed session flag: \n${JSON.stringify(request, null, 2)}`,
-        );
-
         // TODO: Set requestHasReadOrQueryAction flag here when Read/Query are implemented.
-
         try {
           const reqStatus = this.clientExecutor.startHandlingRequest(
             request,

handwritten/spanner/google-cloud-spanner-executor/src/worker-proxy.ts

@@ -42,8 +42,7 @@ const OPTION_CERTIFICATE = 'cert';
 const OPTION_SERVICE_KEY_FILE = 'service_key_file';
 const OPTION_USE_PLAIN_TEXT_CHANNEL = 'use_plain_text_channel';
 const OPTION_ENABLE_GRPC_FAULT_INJECTOR = 'enable_grpc_fault_injector';
-const OPTION_MULTIPLEXED_SESSION_OPERATIONS_RATIO =
-  'multiplexed_session_operations_ratio';
+
 
 /**
  * Acts as a proxy server that forwards incoming gRPC requests to the underlying
@@ -54,7 +53,7 @@ export class WorkerProxy {
   public static proxyPort = 0;
   public static cert = '';
   public static serviceKeyFile = '';
-  public static multiplexedSessionOperationsRatio = 0.0;
+
   public static usePlainTextChannel = false;
   public static enableGrpcFaultInjector = false;
   public static openTelemetrySdk: NodeTracerProvider;
@@ -126,10 +125,7 @@ export class WorkerProxy {
       type: 'boolean',
       description: 'Enable grpc fault injector in cloud client executor.',
     });
-    parser.option(OPTION_MULTIPLEXED_SESSION_OPERATIONS_RATIO, {
-      type: 'number',
-      description: 'Ratio of operations to use multiplexed sessions.',
-    });
+
 
     try {
       return parser.parseSync();