@@ -0,0 +1,2779 @@
+/*%0A *%0A *%0A *  Licensed to the Apache Software Foundation (ASF) under one or more%0A *%0A *  contributor license agreements. See the NOTICE file distributed with%0A *%0A *  this work for additional information regarding copyright ownership.%0A *%0A *  The ASF licenses this file to You under the Apache License, Version 2.0%0A *%0A *  (the %22License%22); you may not use this file except in compliance with%0A *%0A *  the License. You may obtain a copy of the License at%0A *%0A *%0A *%0A *  http://www.apache.org/licenses/LICENSE-2.0%0A *%0A *%0A *%0A *  Unless required by applicable law or agreed to in writing, software%0A *%0A *  distributed under the License is distributed on an %22AS IS%22 BASIS,%0A *%0A *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.%0A *%0A *  See the License for the specific language governing permissions and%0A *%0A *  limitations under the License.%0A *%0A * /%0A */%0A%0Apackage org.apache.kylin.streaming.kafka;%0A%0Aimport com.google.common.collect.Lists;%0Aimport org.slf4j.Logger;%0Aimport org.slf4j.LoggerFactory;%0A%0Aimport java.util.List;%0Aimport java.util.concurrent.BlockingQueue;%0A%0A/**%0A * Created by qianzhou on 2/17/15.%0A */%0Apublic abstract class StreamBuilder implements Runnable %7B%0A%0A    private List%3CBlockingQueue%3CStream%3E%3E streamQueues;%0A    private static final Logger logger = LoggerFactory.getLogger(StreamBuilder.class);%0A    private final int batchBuildCount;%0A%0A    public StreamBuilder(List%3CBlockingQueue%3CStream%3E%3E streamQueues, int batchBuildCount) %7B%0A        this.streamQueues = streamQueues;%0A        this.batchBuildCount = batchBuildCount;%0A    %7D%0A%0A%0A    private int getEarliestStreamIndex(Stream%5B%5D streamHead) %7B%0A        long ts = Long.MAX_VALUE;%0A        int idx = 0;%0A        for (int i = 0; i %3C streamHead.length; i++) %7B%0A            if (streamHead%5Bi%5D.getTimestamp() %3C ts) %7B%0A                ts = streamHead%5Bi%5D.getTimestamp();%0A                idx = i;%0A            %7D%0A        %7D%0A        return idx;%0A    %7D%0A%0A    protected abstract void build(List%3CStream%3E streamsToBuild);%0A%0A    @Override%0A    public void run() %7B%0A        try %7B%0A            Stream%5B%5D streamHead = new Stream%5BstreamQueues.size()%5D;%0A            for (int i = 0; i %3C streamQueues.size(); i++) %7B%0A                streamHead%5Bi%5D = streamQueues.get(i).take();%0A            %7D%0A            List%3CStream%3E streamToBuild = Lists.newArrayListWithCapacity(batchBuildCount);%0A            while (true) %7B%0A                if (streamToBuild.size() %3E= batchBuildCount) %7B%0A                    build(streamToBuild);%0A                    streamToBuild.clear();%0A                %7D%0A                int idx = getEarliestStreamIndex(streamHead);%0A                streamToBuild.add(streamHead%5Bidx%5D);%0A                streamHead%5Bidx%5D = streamQueues.get(idx).take();%0A            %7D%0A        %7D catch (InterruptedException e) %7B%0A            logger.error(%22%22, e);%0A        %7D%0A    %7D%0A%7D%0A

add stream builder