IntrinsicMapTaskExecutorFactory implements DataflowMapTaskExecutorFactory { <K> OperationNode createPartialGroupByKeyOperation( Network<Node, Edge> network, ParallelInstructionNode node, PipelineOptions options, DataflowExecutionContext<?> executionContext, DataflowOperationContext operationContext) throws Exception { ParallelInstruction instruction = node.getParallelInstruction(); PartialGroupByKeyInstruction pgbk = instruction.getPartialGroupByKey(); OutputReceiver[] receivers = getOutputReceivers(network, node); Coder<?> windowedCoder = CloudObjects.coderFromCloudObject(CloudObject.fromSpec(pgbk.getInputElementCodec())); if (!(windowedCoder instanceof WindowedValueCoder)) { throw new IllegalArgumentException( String.format( "unexpected kind of input coder for PartialGroupByKeyOperation: %s", windowedCoder)); } Coder<?> elemCoder = ((WindowedValueCoder<?>) windowedCoder).getValueCoder(); if (!(elemCoder instanceof KvCoder)) { throw new IllegalArgumentException( String.format( "unexpected kind of input element coder for PartialGroupByKeyOperation: %s", elemCoder)); } @SuppressWarnings("unchecked") KvCoder<K, ?> keyedElementCoder = (KvCoder<K, ?>) elemCoder; CloudObject cloudUserFn = pgbk.getValueCombiningFn() != null ? CloudObject.fromSpec(pgbk.getValueCombiningFn()) : null; ParDoFn fn = PartialGroupByKeyParDoFns.create( options, keyedElementCoder, cloudUserFn, pgbk.getSideInputs(), Arrays.<Receiver>asList(receivers), executionContext, operationContext); return OperationNode.create(new ParDoOperation(fn, receivers, operationContext)); } private  IntrinsicMapTaskExecutorFactory(); static IntrinsicMapTaskExecutorFactory defaultFactory(); @Override DataflowMapTaskExecutor create(
      InstructionRequestHandler instructionRequestHandler,
      GrpcFnServer<GrpcDataService> grpcDataFnServer,
      Endpoints.ApiServiceDescriptor dataApiServiceDescriptor,
      GrpcFnServer<GrpcStateService> grpcStateFnServer,
      MutableNetwork<Node, Edge> network,
      PipelineOptions options,
      String stageName,
      ReaderFactory readerFactory,
      SinkFactory sinkFactory,
      DataflowExecutionContext<?> executionContext,
      CounterSet counterSet,
      IdGenerator idGenerator);  }
@Test public void testCreatePartialGroupByKeyOperation() throws Exception { int producerIndex = 1; int producerOutputNum = 2; ParallelInstructionNode instructionNode = ParallelInstructionNode.create( createPartialGroupByKeyInstruction(producerIndex, producerOutputNum), ExecutionLocation.UNKNOWN); when(network.successors(instructionNode)) .thenReturn( ImmutableSet.<Node>of( IntrinsicMapTaskExecutorFactory.createOutputReceiversTransform(STAGE, counterSet) .apply( InstructionOutputNode.create( instructionNode.getParallelInstruction().getOutputs().get(0), PCOLLECTION_ID)))); when(network.outDegree(instructionNode)).thenReturn(1); Node operationNode = mapTaskExecutorFactory .createOperationTransformForParallelInstructionNodes( STAGE, network, PipelineOptionsFactory.create(), readerRegistry, sinkRegistry, BatchModeExecutionContext.forTesting(options, counterSet, "testStage")) .apply(instructionNode); assertThat(operationNode, instanceOf(OperationNode.class)); assertThat(((OperationNode) operationNode).getOperation(), instanceOf(ParDoOperation.class)); ParDoOperation pgbkOperation = (ParDoOperation) ((OperationNode) operationNode).getOperation(); assertEquals(1, pgbkOperation.receivers.length); assertEquals(0, pgbkOperation.receivers[0].getReceiverCount()); assertEquals(Operation.InitializationState.UNSTARTED, pgbkOperation.initializationState); }