import storm.trident.JoinType; //导入依赖的package包/类
@Override
public void complete(JoinState state, TridentTuple group, TridentCollector collector) {
    List<List>[] sides = state.sides;
    boolean wasEmpty = state.numSidesReceived < sides.length;
    for(int i=0; i<sides.length; i++) {
        if(sides[i].isEmpty() && _types.get(i) == JoinType.OUTER) {
            state.numSidesReceived++;
            sides[i].add(makeNullList(_sideFields.get(i).size()));
        }
    }
    if(wasEmpty && state.numSidesReceived == sides.length) {
        emitCrossJoin(state, collector, -1, null);
    }
}
 

import storm.trident.JoinType; //导入依赖的package包/类
@Override
public void complete(JoinState state, TridentTuple group, TridentCollector collector) {
    List<List>[] sides = state.sides;
    boolean wasEmpty = state.numSidesReceived < sides.length;
    for (int i = 0; i < sides.length; i++) {
        if (sides[i].isEmpty() && _types.get(i) == JoinType.OUTER) {
            state.numSidesReceived++;
            sides[i].add(makeNullList(_sideFields.get(i).size()));
        }
    }
    if (wasEmpty && state.numSidesReceived == sides.length) {
        emitCrossJoin(state, collector, -1, null);
    }
}
 

import storm.trident.JoinType; //导入依赖的package包/类
public JoinerMultiReducer(List<JoinType> types, int numGroupFields, List<Fields> sides) {
    _types = types;
    _sideFields = sides;
    _numGroupFields = numGroupFields;
}