import org.apache.commons.collections.list.UnmodifiableList; //导入依赖的package包/类
@Override
public <K, V> Collection<ConsumerRecord<K, V>> getAllRecordsFromTopic(Properties properties, String topic, int partition, long timeout) {
	List<ConsumerRecord<K, V>> result = new ArrayList<>();
	try (KafkaConsumer<K, V> consumer = new KafkaConsumer<>(properties)) {
		consumer.assign(Arrays.asList(new TopicPartition(topic, partition)));

		while (true) {
			boolean processedAtLeastOneRecord = false;

			Iterator<ConsumerRecord<K, V>> iterator = consumer.poll(timeout).iterator();
			while (iterator.hasNext()) {
				ConsumerRecord<K, V> record = iterator.next();
				result.add(record);
				processedAtLeastOneRecord = true;
			}

			if (!processedAtLeastOneRecord) {
				break;
			}
		}
		consumer.commitSync();
	}

	return UnmodifiableList.decorate(result);
}
 

import org.apache.commons.collections.list.UnmodifiableList; //导入依赖的package包/类
@Override
public <K, V> Collection<ConsumerRecord<K, V>> getAllRecordsFromTopic(Properties properties, String topic, int partition, long timeout) {
	List<ConsumerRecord<K, V>> result = new ArrayList<>();
	try (KafkaConsumer<K, V> consumer = new KafkaConsumer<>(properties)) {
		consumer.subscribe(new TopicPartition(topic, partition));

		while (true) {
			Map<String, ConsumerRecords<K, V>> topics = consumer.poll(timeout);
			if (topics == null || !topics.containsKey(topic)) {
				break;
			}
			List<ConsumerRecord<K, V>> records = topics.get(topic).records(partition);
			result.addAll(records);
			if (records.size() == 0) {
				break;
			}
		}
		consumer.commit(true);
	}

	return UnmodifiableList.decorate(result);
}
 

import org.apache.commons.collections.list.UnmodifiableList; //导入依赖的package包/类
@Override
public <K, V> Collection<ConsumerRecord<K, V>> getAllRecordsFromTopic(Properties properties, String topic, int partition, long timeout) {
	List<ConsumerRecord<K, V>> result = new ArrayList<>();

	try (KafkaConsumer<K, V> consumer = new KafkaConsumer<>(properties)) {
		consumer.assign(Arrays.asList(new TopicPartition(topic, partition)));

		while (true) {
			boolean processedAtLeastOneRecord = false;

			// wait for new records with timeout and break the loop if we didn't get any
			Iterator<ConsumerRecord<K, V>> iterator = consumer.poll(timeout).iterator();
			while (iterator.hasNext()) {
				ConsumerRecord<K, V> record = iterator.next();
				result.add(record);
				processedAtLeastOneRecord = true;
			}

			if (!processedAtLeastOneRecord) {
				break;
			}
		}
		consumer.commitSync();
	}

	return UnmodifiableList.decorate(result);
}
 

import org.apache.commons.collections.list.UnmodifiableList; //导入依赖的package包/类
/**
 * Retrieves all the components of this name.
 *
 * @return All the components
 */
@SuppressWarnings("unchecked")
public List<Rdn> getRdns()
{
    return UnmodifiableList.decorate( rdns );
}
 

import org.apache.commons.collections.list.UnmodifiableList; //导入依赖的package包/类
/**
 * Returns the configuration of this instance - implemented for testing purpose only
 * @return
 */
@SuppressWarnings("unchecked")
protected List<ProjectionMapperConfiguration> getConfiguration() {
	return UnmodifiableList.decorate(this.configuration);
}
 

import org.apache.commons.collections.list.UnmodifiableList; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public List<String> getColumnNames() {
	return UnmodifiableList.decorate(simpleColumnNames);
}
 

import org.apache.commons.collections.list.UnmodifiableList; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public List<UserRight> getOrderedRights()
{
  return UnmodifiableList.decorate(orderedRights);
}
 

import org.apache.commons.collections.list.UnmodifiableList; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public List<String> getList() {
    return UnmodifiableList.decorate(_list);
}