import avro.shaded.com.google.common.collect.Lists; //导入依赖的package包/类
/**
 * Divide the part files into the batches to download and pre-sort at the same time.
 *
 * @param partFiles all part files to download.
 *
 * @return a map of path to download the parts and the parts that should be used (as reads
 * source).
 *
 * @see #downloadBatchesAndPreSort(List)
 */
private Map<Path, ReadsDataSource> divideIntoBatches(final List<Path> partFiles) {
    // partition the files into batches
    final List<List<Path>> batches = Lists.partition(partFiles, numberOfParts);
    final Map<Path, ReadsDataSource> toReturn = new LinkedHashMap<>(batches.size());

    // creates a temp file for each in a common temp folder
    final File tempDir = IOUtil.createTempDir(this.toString(), ".batches");
    int i = 0;
    for (final List<Path> parts : batches) {
        // create a temp file and store it in the temp parts
        final Path tempFile =
                IOUtils.getPath(new File(tempDir, "batch-" + i++ + ".bam").getAbsolutePath());
        logger.debug("Batch file {} will contain {} parts: {}",
                () -> tempFile.toUri().toString(),
                () -> parts.size(),
                () -> parts.stream().map(p -> p.toUri().toString())
                        .collect(Collectors.toList()));
        toReturn.put(tempFile, new ReadsDataSource(parts, getSamReaderFactory()));
    }

    return toReturn;
}
 

import avro.shaded.com.google.common.collect.Lists; //导入依赖的package包/类
private static List<String> toListOfString(Map<Object, ?> map) {
    List<Object> keysList = map.keySet().stream().collect(Collectors.toList());
    List<String> keysListString = Lists.newArrayList();
    keysList.forEach(key -> keysListString.add(String.valueOf(key)));

    return keysListString;
}
 

import avro.shaded.com.google.common.collect.Lists; //导入依赖的package包/类
private Consumer<String, byte[]> createConsumer() throws Exception {
    Properties props = PropertiesHolder.getProperties(Constants.Properties.CONSUMER_CONFIG);

    props.setProperty("client.id", this.topologyId + "_consumer");
    props.setProperty("group.id", this.topologyId + "_grp");

    TopicProvider dataTopicProvider = new DataInputTopicProvider();
    TopicProvider controlTopicProvider = new ControlTopicProvider();
    TopicProvider schemaTopicProvider = new SchemaTopicProvider();
    this.controlTopics = controlTopicProvider.provideTopics();
    this.dataTopics = dataTopicProvider.provideTopics();

    List<String> topics = Lists.newArrayList();
    topics.addAll(controlTopics);
    topics.addAll(dataTopics);

    if(DbusDatasourceType.ORACLE == GlobalCache.getDatasourceType()) {
    	this.schemaTopics = schemaTopicProvider.provideTopics();
     topics.addAll(schemaTopics);
    }
    Consumer<String, byte[]> consumer = consumerProvider.consumer(props, topics);

    Map<String, Object> data = zkNodeOperator.getData();
    for (Map.Entry<String, Object> entry : data.entrySet()) {
        TopicInfo b = TopicInfo.build((Map<String, Object>) entry.getValue());
        this.pausedTopics.put(b.getTopic(), b);
    }

    consumerProvider.pause(consumer, this.pausedTopics.entrySet()
            .stream()
            .map(entry -> entry.getValue())
            .collect(Collectors.toList()));

    return consumer;
}
 

import avro.shaded.com.google.common.collect.Lists; //导入依赖的package包/类
/**
 * Get the step from the hierachical string path
 *
 * @param path
 * @return
 */
public static Stack<String> getPathSteps(String path) {
    if (path.startsWith(".")) {
        path = path.substring(1);
    }
    Stack<String> pathSteps = new Stack<String>();
    List<String> stepsList = Arrays.asList(path.split("\\."));
    pathSteps.addAll(Lists.reverse(stepsList));
    return pathSteps;
}
 

import avro.shaded.com.google.common.collect.Lists; //导入依赖的package包/类
@Test
public void testGetDeltaFieldNamesForNewSchema(){
  Configuration conf = mock(Configuration.class);
  when(conf.get(FieldAttributeBasedDeltaFieldsProvider.ATTRIBUTE_FIELD)).thenReturn("attributes_json");
  when(conf.get(FieldAttributeBasedDeltaFieldsProvider.DELTA_PROP_NAME,
      FieldAttributeBasedDeltaFieldsProvider.DEFAULT_DELTA_PROP_NAME))
      .thenReturn(FieldAttributeBasedDeltaFieldsProvider.DEFAULT_DELTA_PROP_NAME);

  AvroDeltaFieldNameProvider provider = new FieldAttributeBasedDeltaFieldsProvider(conf);
  Schema original = new Schema.Parser().parse(FULL_SCHEMA_WITH_ATTRIBUTES);
  GenericRecord record = mock(GenericRecord.class);
  when(record.getSchema()).thenReturn(original);
  List<String> fields = provider.getDeltaFieldNames(record);
  Assert.assertEquals(fields, Lists.newArrayList("scn2", "scn"));
}
 

import avro.shaded.com.google.common.collect.Lists; //导入依赖的package包/类
/**
 *
 * @param reloadJson: reload control msg json
 */
private void reloadConfig(String reloadJson) {

    close();

    ZKHelper zkHelper = null;
    DBHelper dbHelper = null;
    try {
        dsInfo = new DataSourceInfo();

        zkHelper = new ZKHelper(topologyRoot, topologyID, zkServers);
        zkHelper.loadDsNameAndOffset(dsInfo);

        dbHelper = new DBHelper(zkServers);
        dbHelper.loadDsInfo(dsInfo);
        logger.info(String.format("Spout read datasource: %s", dsInfo.toString()));

        //init consumer
        dataTopicPartition = new TopicPartition(dsInfo.getDataTopic(), 0);
        ctrlTopicPartition = new TopicPartition(dsInfo.getCtrlTopic(), 0);
        List<TopicPartition> topics = Arrays.asList(dataTopicPartition, ctrlTopicPartition);
        consumer = new KafkaConsumer(zkHelper.getConsumerProps());
        consumer.assign(topics);

        //skip offset
        long oldOffset = consumer.position(dataTopicPartition);
        logger.info(String.format("reloaded offset as: %d", oldOffset));

        String offset = dsInfo.getDataTopicOffset();
        if (offset.equalsIgnoreCase("none")) {
            ; // do nothing

        } else if (offset.equalsIgnoreCase("begin")) {
            consumer.seekToBeginning(Lists.newArrayList(dataTopicPartition));
            logger.info(String.format("Offset seek to begin, changed as: %d", consumer.position(dataTopicPartition)));

        } else if (offset.equalsIgnoreCase("end")) {
            consumer.seekToEnd(Lists.newArrayList(dataTopicPartition));
            logger.info(String.format("Offset seek to end, changed as: %d", consumer.position(dataTopicPartition)));
        } else {
            long nOffset = Long.parseLong(offset);
            consumer.seek(dataTopicPartition, nOffset);
            logger.info(String.format("Offset changed as: %d", consumer.position(dataTopicPartition)));
        }
        dsInfo.resetDataTopicOffset();

        zkHelper.saveDsInfo(dsInfo);
        zkHelper.saveReloadStatus(reloadJson, "dispatcher-spout", true);

    } catch (Exception ex) {
        logger.error("KafkaConsumerSpout reloadConfig():", ex);
        collector.reportError(ex);
        throw new RuntimeException(ex);
    } finally {
        if (dbHelper != null) {
            dbHelper.close();
        }
        if (zkHelper != null) {
            zkHelper.close();
        }
    }
}