我们正在尝试实现kafka作为我们的消息代理解决方案。我们正在ibmbluemix中部署spring-boot微服务，其内部消息代理实现是kafka版本0.10。由于我的经验更多地是在jms、activemq端，我想知道什么是处理java用户中的系统级错误的理想方法？
下面是我们目前是如何实现的
消费者财产

enable.auto.commit=false
auto.offset.reset=latest

我们正在使用的默认属性

max.partition.fetch.bytes
session.timeout.ms

Kafka消费者
我们正在为每个主题旋转3个线程，所有线程都具有相同的groupid，即每个线程一个kafkaconsumer示例。我们现在只有一个分区。在thread类的构造函数中，使用者代码如下所示

kafkaConsumer = new KafkaConsumer<String, String>(properties);

    final List<String> topicList = new ArrayList<String>();
    topicList.add(properties.getTopic());

    kafkaConsumer.subscribe(topicList, new ConsumerRebalanceListener() {

        @Override
        public void onPartitionsRevoked(final Collection<TopicPartition> partitions) {
        }

        @Override
        public void onPartitionsAssigned(final Collection<TopicPartition> partitions) {
            try {
                logger.info("Partitions assigned, consumer seeking to end.");

                for (final TopicPartition partition : partitions) {
                    final long position = kafkaConsumer.position(partition);
                    logger.info("current Position: " + position);

                    logger.info("Seeking to end...");
                    kafkaConsumer.seekToEnd(Arrays.asList(partition));
                    logger.info("Seek from the current position: " + kafkaConsumer.position(partition));
                    kafkaConsumer.seek(partition, position);
                }
                logger.info("Consumer can now begin consuming messages.");
            } catch (final Exception e) {
                logger.error("Consumer can now begin consuming messages.");
            }

        }
    });

实际读取发生在线程的run方法中

try {
            // Poll on the Kafka consumer every second.
            final ConsumerRecords<String, String> records = kafkaConsumer.poll(1000);

            // Iterate through all the messages received and print their
            // content.
            for (final TopicPartition partition : records.partitions()) {

                final List<ConsumerRecord<String, String>> partitionRecords = records.records(partition);
                logger.info("consumer is alive and is processing   "+ partitionRecords.size() +" records");
                for (final ConsumerRecord<String, String> record : partitionRecords) {
                    logger.info("processing topic  "+ record.topic()+" for key "+record.key()+" on offset "+ record.offset());

                    final Class<? extends Event> resourceClass = eventProcessors.getResourceClass();
                    final Object obj = converter.convertToObject(record.value(), resourceClass);
                    if (obj != null) {
                        logger.info("Event: " + obj + " acquired by  " + Thread.currentThread().getName());
                        final CommsEvent event = resourceClass.cast(converter.convertToObject(record.value(), resourceClass));
                        final MessageResults results = eventProcessors.processEvent(event
                                );
                        if ("Success".equals(results.getStatus())) {
                            // commit the processed message which changes
                            // the offset
                            kafkaConsumer.commitSync();
                            logger.info("Message processed sucessfully");
                        } else {
                            kafkaConsumer.seek(new TopicPartition(record.topic(), record.partition()), record.offset());
                            logger.error("Error processing message : {} with error : {},resetting offset to {} ", obj,results.getError().getMessage(),record.offset());
                            break;
                        }

                    }
                }

            }
            // TODO add return

        } catch (final Exception e) {
            logger.error("Consumer has failed with exception: " + e, e);
            shutdown();
        }

您将注意到eventprocessor，它是一个处理每个记录的服务类，在大多数情况下提交数据库中的记录。如果处理器抛出错误（系统异常或validationexception），我们不会提交，而是通过编程将seek设置为该偏移量，以便后续轮询将从该偏移量返回该组id。
现在的疑问是，这是正确的方法吗？如果我们得到一个错误并设置了偏移量，那么在修复之前，不会处理其他消息。这可能适用于系统错误，如无法连接到数据库，但如果问题只是与该事件，而不是其他人处理这一个记录，我们将无法处理任何其他记录。我们想到了errortopic的概念，当我们得到一个错误时，消费者会将该事件发布到errortopic，同时它会继续处理其他后续事件。但是看起来我们正在尝试将jms的设计概念（由于我以前的经验）引入kafka，并且可能有更好的方法来解决kafka中的错误处理。另外，从错误主题重新处理它可能会改变某些场景中不需要的消息序列
请让我知道有人如何处理这个场景在他们的项目遵循Kafka标准。
-塔塔