小编col*_*sal的帖子

我正在使用KafkaConsumer 0.10 Java api.我想从特定的分区和特定的偏移量消耗.我查了一下,发现有一个搜索方法,但它抛出异常.任何人都有类似的用例或解决方案？

码:

KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(consumerProps);
consumer.seek(new TopicPartition("mytopic", 1), 4);

例外

java.lang.IllegalStateException: No current assignment for partition mytopic-1
    at org.apache.kafka.clients.consumer.internals.SubscriptionState.assignedState(SubscriptionState.java:251)
    at org.apache.kafka.clients.consumer.internals.SubscriptionState.seek(SubscriptionState.java:276)
    at org.apache.kafka.clients.consumer.KafkaConsumer.seek(KafkaConsumer.java:1135)
    at xx.xxx.xxx.Test.main(Test.java:182)

我正在使用java 8流API对Store对象列表执行操作.

Store接受一个String论点和一个Mapper对象.Mapper对所有Store对象都是一样的.

问:我如何传递Mapper对象时我初始化Store这里.map(Store::new)？

public class Store {
    public Store(String name, Mapper mapper) {
    }
}

public class Mapper {
}

public class Test {
    public static void main(String[] args) {
        List<String> names = new ArrayList<String>();

        Mapper mapper = new Mapper();
         // compile time problem at Store::new because it takes 2 arguments
         List<Store> actions = 
             names.stream()
              .map(Store::new)
              .collect(Collectors.toList());
    }
}