String sinkName = sink.getName();
String sourceClassName = sourceClassMap.get(sourceName);
String sinkClassName = sinkClassMap.get(sinkName);
BatchSource batchSource = (BatchSource) Class.forName(sourceClassName).newInstance();
BatchSink batchSink = (BatchSink) Class.forName(sinkClassName).newInstance();
<BUG>if (transform.size() == 0) {
Preconditions.checkArgument(batchSink.getKeyType().getClass().isAssignableFrom(
batchSource.getKeyType().getClass()));
Preconditions.checkArgument(batchSink.getValueType().getClass().isAssignableFrom(
batchSource.getValueType().getClass()));</BUG>
} else {

String sinkName = sink.getName();
String sourceClassName = sourceClassMap.get(sourceName);
String sinkClassName = sinkClassMap.get(sinkName);
BatchSource batchSource = (BatchSource) Class.forName(sourceClassName).newInstance();
BatchSink batchSink = (BatchSink) Class.forName(sinkClassName).newInstance();
if (transform.size() == 0) {
Preconditions.checkArgument(batchSink.getKeyType().equals(batchSource.getKeyType()));
Preconditions.checkArgument(batchSink.getValueType().equals(batchSource.getValueType()));
} else {
