public static SingleInputGate create(
		String owningTaskName,
		JobID jobId,
		InputGateDeploymentDescriptor igdd,
		NetworkEnvironment networkEnvironment,
		TaskEventPublisher taskEventPublisher,
		TaskActions taskActions,
		InputChannelMetrics metrics,
		Counter numBytesInCounter) {

		final IntermediateDataSetID consumedResultId = checkNotNull(igdd.getConsumedResultId());
		final ResultPartitionType consumedPartitionType = checkNotNull(igdd.getConsumedPartitionType());

		final int consumedSubpartitionIndex = igdd.getConsumedSubpartitionIndex();
		checkArgument(consumedSubpartitionIndex >= 0);

		final InputChannelDeploymentDescriptor[] icdd = checkNotNull(igdd.getInputChannelDeploymentDescriptors());

		final NetworkEnvironmentConfiguration networkConfig = networkEnvironment.getConfiguration();

		final SingleInputGate inputGate = new SingleInputGate(
			owningTaskName, jobId, consumedResultId, consumedPartitionType, consumedSubpartitionIndex,
			icdd.length, taskActions, numBytesInCounter, networkConfig.isCreditBased());

		// Create the input channels. There is one input channel for each consumed partition.
		final InputChannel[] inputChannels = new InputChannel[icdd.length];

		int numLocalChannels = 0;
		int numRemoteChannels = 0;
		int numUnknownChannels = 0;

		for (int i = 0; i < inputChannels.length; i++) {
			final ResultPartitionID partitionId = icdd[i].getConsumedPartitionId();
			final ResultPartitionLocation partitionLocation = icdd[i].getConsumedPartitionLocation();

			if (partitionLocation.isLocal()) {
				inputChannels[i] = new LocalInputChannel(inputGate, i, partitionId,
					networkEnvironment.getResultPartitionManager(),
					taskEventPublisher,
					networkConfig.partitionRequestInitialBackoff(),
					networkConfig.partitionRequestMaxBackoff(),
					metrics
				);

				numLocalChannels++;
			}
			else if (partitionLocation.isRemote()) {
				inputChannels[i] = new RemoteInputChannel(inputGate, i, partitionId,
					partitionLocation.getConnectionId(),
					networkEnvironment.getConnectionManager(),
					networkConfig.partitionRequestInitialBackoff(),
					networkConfig.partitionRequestMaxBackoff(),
					metrics
				);

				numRemoteChannels++;
			}
			else if (partitionLocation.isUnknown()) {
				inputChannels[i] = new UnknownInputChannel(inputGate, i, partitionId,
					networkEnvironment.getResultPartitionManager(),
					taskEventPublisher,
					networkEnvironment.getConnectionManager(),
					networkConfig.partitionRequestInitialBackoff(),
					networkConfig.partitionRequestMaxBackoff(),
					metrics
				);

				numUnknownChannels++;
			}
			else {
				throw new IllegalStateException("Unexpected partition location.");
			}

			inputGate.setInputChannel(partitionId.getPartitionId(), inputChannels[i]);
		}

		LOG.debug("{}: Created {} input channels (local: {}, remote: {}, unknown: {}).",
			owningTaskName,
			inputChannels.length,
			numLocalChannels,
			numRemoteChannels,
			numUnknownChannels);

		return inputGate;
	}
public static SingleInputGate create ( String owningTaskName , JobID jobId , InputGateDeploymentDescriptor igdd , NetworkEnvironment networkEnvironment , TaskEventPublisher taskEventPublisher , TaskActions taskActions , InputChannelMetrics metrics , Counter numBytesInCounter ) { final IntermediateDataSetID consumedResultId = checkNotNull ( igdd . getConsumedResultId ( ) ) ; final ResultPartitionType consumedPartitionType = checkNotNull ( igdd . getConsumedPartitionType ( ) ) ; final int consumedSubpartitionIndex = igdd . getConsumedSubpartitionIndex ( ) ; checkArgument ( consumedSubpartitionIndex >= 0 ) ; final InputChannelDeploymentDescriptor [ ] icdd = checkNotNull ( igdd . getInputChannelDeploymentDescriptors ( ) ) ; final NetworkEnvironmentConfiguration networkConfig = networkEnvironment . getConfiguration ( ) ; final SingleInputGate inputGate = new SingleInputGate ( owningTaskName , jobId , consumedResultId , consumedPartitionType , consumedSubpartitionIndex , icdd . length , taskActions , numBytesInCounter , networkConfig . isCreditBased ( ) ) ; final InputChannel [ ] inputChannels = new InputChannel [ icdd . length ] ; int numLocalChannels = 0 ; int numRemoteChannels = 0 ; int numUnknownChannels = 0 ; for ( int i = 0 ; i < inputChannels . length ; i ++ ) { final ResultPartitionID partitionId = icdd [ i ] . getConsumedPartitionId ( ) ; final ResultPartitionLocation partitionLocation = icdd [ i ] . getConsumedPartitionLocation ( ) ; if ( partitionLocation . isLocal ( ) ) { inputChannels [ i ] = new LocalInputChannel ( inputGate , i , partitionId , networkEnvironment . getResultPartitionManager ( ) , taskEventPublisher , networkConfig . partitionRequestInitialBackoff ( ) , networkConfig . partitionRequestMaxBackoff ( ) , metrics ) ; numLocalChannels ++ ; } else if ( partitionLocation . isRemote ( ) ) { inputChannels [ i ] = new RemoteInputChannel ( inputGate , i , partitionId , partitionLocation . getConnectionId ( ) , networkEnvironment . getConnectionManager ( ) , networkConfig . partitionRequestInitialBackoff ( ) , networkConfig . partitionRequestMaxBackoff ( ) , metrics ) ; numRemoteChannels ++ ; } else if ( partitionLocation . isUnknown ( ) ) { inputChannels [ i ] = new UnknownInputChannel ( inputGate , i , partitionId , networkEnvironment . getResultPartitionManager ( ) , taskEventPublisher , networkEnvironment . getConnectionManager ( ) , networkConfig . partitionRequestInitialBackoff ( ) , networkConfig . partitionRequestMaxBackoff ( ) , metrics ) ; numUnknownChannels ++ ; } else { throw new IllegalStateException ( "Unexpected partition location." ) ; } inputGate . setInputChannel ( partitionId . getPartitionId ( ) , inputChannels [ i ] ) ; } LOG . debug ( "{}: Created {} input channels (local: {}, remote: {}, unknown: {})." , owningTaskName , inputChannels . length , numLocalChannels , numRemoteChannels , numUnknownChannels ) ; return inputGate ; }
Creates an input gate and all of its input channels.