인프런 커뮤니티 질문&답변

선생님 안녕하세요.

죄송합니다. 잘 모르다보니 횡설수설 했습니다. 생각을 좀 정리하였고, 다시 한 번 말씀드리겠습니다.

조건 : seek(N)으로 데이터를 읽으려고 할 때 토픽에 N보다 적은 수의 데이터가 있는 경우 (즉 offset이 없거나 N보다 작을 경우)

현상 : 처음 FetchPosition의 offset은 N이었으나 이어서 consumer가 토픽의 마지막 offset으로 컨슈머의 FetchPosition을 덮어쓰고, topic에 데이터가 쌓이면 그 설정된 FetchPosition(토픽의 마지막 offset) 데이터를 읽어옵니다 (Fetcher.java의 resetOffsetIfNeeded() 에서 수행됩니다. )

왜 FetchPosition이 seek(N)했을 때 FetchPosition{offset=N}이 어느 순간 토픽의 마지막 offset으로 덮어씌워지는 정책을 가지는지 궁금합니다.

(FetchPosition의 offset을 seek에서 준 값을 그대로 가지고 있지, 왜 굳이 토픽의 마지막 오프셋으로 업데이트하지? 싶어서요)

// seek 값 10을 가지는 FetchPosition 로그 (FetchPosition{offset=10 ...)
[main] INFO org.apache.kafka.clients.consumer.internals.Fetcher - [Consumer clientId=consumer-group-pizza-assign-seek-maintenance-11-1, groupId=group-pizza-assign-seek-maintenance-11] Fetch position FetchPosition{offset=10, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[localhost:9092 (id: 0 rack: null)], epoch=0}} is out of range for partition pizza-topic-0, resetting offset

// topic의 마지막 offset인 1로 덮어씌워지는 FetchPosition 로그 (FetchPosition{offset=1 ...)
[main] INFO org.apache.kafka.clients.consumer.internals.SubscriptionState - [Consumer clientId=consumer-group-pizza-assign-seek-maintenance-11-1, groupId=group-pizza-assign-seek-maintenance-11] Resetting offset for partition pizza-topic-0 to position FetchPosition{offset=1, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[localhost:9092 (id: 0 rack: null)], epoch=0}}.

테스트한 코드도 첨부드립니다.

(partiton3개인 pizzaProducer를 사용하였고 consumer도 선생님께서 강의에서 진행해주신 코드와 거의 똑같습니다)

  public static void main(String[] args) {

        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "group-pizza-assign-seek-maintenance-11");
        props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        String topicName = "pizza-topic";

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

        TopicPartition topicPartition = new TopicPartition(topicName, 0);

        kafkaConsumer.assign(List.of(topicPartition));
        kafkaConsumer.seek(topicPartition, 10L);

        Thread mainThread = Thread.currentThread();

        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            log.info("main program starts to exit by calling wakeup");
            kafkaConsumer.wakeup();

            try{
                mainThread.join();
            } catch(InterruptedException e) {
                e.printStackTrace();
            }
        }));


        pollCommitSync(kafkaConsumer);
    }

    private static void pollCommitSync(final KafkaConsumer<String, String> kafkaConsumer) {
        int loopCnt = 0;
        try {
            while(true) {
                ConsumerRecords<String, String> records =  kafkaConsumer.poll(Duration.ofMillis(1_000L));
                log.info("loopCnt: {}, consumerRecords count: {}", loopCnt++, records.count());
                for(ConsumerRecord<String, String> record: records) {
                    log.info("record key: {}, partition: {}, recordOffset: {}",
                        record.key(), record.partition(), record.offset());
                }
                
                try {
                    if (records.count() > 100 ){
                        kafkaConsumer.commitSync();
                        log.info("commit sync called");
                    }
                } catch(CommitFailedException e) {
                    log.error("CommitFailedException! ", e);
                }
            }
        } catch (WakeupException e) {
            log.error("wakeup exception has been called", e);
        } finally {
            log.info("finally consumer is closed");
            kafkaConsumer.close();
        }
    }

질문이 고르지 못했던 점 양해 부탁드립니다.

주말에도 답변 감사합니다 선생님!

오, 굉장히 재미있는(?) 부분을 테스트 해보셨군요.

seek(N)인데 아직 offset이 N까지 없는 경우 , (제 기억으로는) 과거 버전에서는 Out of range 오류를 내었던 걸로 기억합니다만,

근데 지금은 out of range 오류 시 default offset으로 Fetcher가 position을 잡는 걸로 보입니다. 만약 이렇게 처리하지 않으면 오류를 발생 시키던지, 아님 N 이 될 때 까지 Consumer가 계속 대기 해야 하는데, 이런 부분을 개선하기 위해서 default offset으로 resetting하는 걸로 추정됩니다.

아하!

이런 현상이 카프카 설계 원리로 인해 발생하는 것이 아니라,

현 버전의 정책정도일 뿐이고 계속해서 업데이트 될 수 있는 부분이군요!

답변 감사드립니다 선생님. 남은 강의도 열심히 듣겠습니다!

즐거운 한주 시작하십쇼!