이용자 특성 구성

소비자를 위해 동작의 여러 측면을 제어하는 많은 구성 설정이 존재합니다. 다음 설정은 가장 중요한 설정 중 일부입니다.

더 많은 구성 설정을 사용할 수 있지만 실험을 시작하기 전에 Apache Kafka 문서 를 읽으십시오.

이용자 그룹

_이용자 그룹_은 하나 이상의 토픽에서 메시지를 이용하도록 협력하는 이용자 그룹입니다. 그룹의 이용자는 모두 동일한 값의 group.id 구성을 사용합니다. 워크로드를 처리하는 데 둘 이상의 이용자가 필요한 경우 동일한 이용자 그룹에서 여러 이용자를 실행할 수 있습니다. 하나의 이용자만 필요한 경우에도 group.id의 값을 지정하는 것이 일반적입니다.

각 이용자 그룹에는 그룹의 이용자에게 파티션을 지정할 책임이 있는 _코디네이터_라는 클러스터의 서버가 있습니다. 이 책임은 로드가 고르게 나뉘도록 클러스터의 서버 전체에 퍼져 있습니다. 이용자에 대한 파티션 지정은 그룹을 리밸런싱할 때마다 변경될 수 있습니다.

이용자는 이용자 그룹에 참여할 때 그룹의 조정자를 검색합니다. 그러면 이용자는 조정자에게 그룹에 참여하고 싶다고 알리고, 조정자는 새 멤버를 포함하도록 그룹 전체에서 파티션의 재조정을 시작합니다.

단일 파티션의 메시지는 각 그룹에서 하나의 이용자만 처리합니다. 이 메커니즘을 사용하면 각 파티션의 메시지를 순서대로 처리할 수 있습니다. 주제에 세 개의 파티션이 포함되어 있고 해당 주제를 이용하는 이용자 그룹에 두 개의 이용자가 포함되어 있는 예제는 다음 다이어그램을 참조하십시오. 그룹의 한 이용자에게는 두 개의 파티션이 지정되고 다른 이용자에게는 한 개의 파티션이 지정됩니다.

이용자 그룹에서 다음 변경사항 중 하나가 발생하면 변경사항을 수용하기 위해 파티션의 지정을 그룹 멤버로 이동하여 그룹을 재조정합니다.

• 이용자가 그룹에 참여합니다.
• 이용자가 그룹에서 나갑니다.
• 조정자는 이용자가 더 이상 상주하지 않는 것으로 간주합니다.
• 새 파티션이 기존 토픽에 추가됩니다.

밸런스가 재조정된 소비자 그룹이 있는 경우 그룹을 떠난 모든 소비자는 그룹에 다시 참여할 때까지 커밋이 거부된다는 점에 유의하세요. 이 경우 이용자는 그룹에 다시 참여해야 하며 이전에 이용한 파티션에 다른 파티션이 지정될 수 있습니다.

이용자 활성 상태(liveness)

Kafka는 작업 중인 이용자에게 파티션을 다시 지정할 수 있도록 실패한 이용자를 자동으로 발견합니다. 폴링 및 하트비트라는 두 가지 메커니즘을 사용합니다.

If the batch of messages that are returned from Consumer.poll(...) is large or the processing is time-consuming, the delay before poll() is called again can be significant or unpredictable. 어떤 경우에는, 메시지 처리에 시간이 걸린다는 이유로 이용자가 그룹에서 제거되지 않도록 최대 폴링 간격을 길게 구성해야 합니다. 이 메커니즘이 유일하게 사용 가능한 메커니즘인 경우 실패한 이용자를 발견하는 데 걸리는 시간도 길어집니다.

이용자 활성 상태(liveness) 확인을 쉽게 처리할 수 있도록 Kafka 0.10.1에 백그라운드 하트비트가 추가되었습니다. 그룹 조정자는 그룹 멤버가 활성임을 표시하기 위해 주기적으로 하트비트를 전송하도록 요청합니다. 백그라운드 하트비트 스레드가 이용자에서 실행되고 일반 하트비트를 조정자로 전송합니다. 조정자가 _세션 제한시간_내에 그룹 멤버로부터 하트비트를 받지 않으면 조정자가 그룹에서 멤버를 제거하고 그룹의 재조정을 시작합니다. 세션 타임아웃은 최대 폴링 간격보다 훨씬 짧을 수 있으므로 메시지 처리에 시간이 오래 걸리더라도 실패한 소비자를 감지하는 데 걸리는 시간이 짧아질 수 있습니다.

max.poll.interval.ms 특성을 사용하여 최대 폴링 간격을 구성하고 session.timeout.ms 특성을 사용하여 세션 제한시간을 구성할 수 있습니다. 메시지 일괄처리를 처리하는 데 5분 넘게 걸리지 않으면 해당 설정을 사용할 필요가 없습니다.

오프셋 관리

Kafka에서는 이용자 그룹마다 이용되는 각 파티션에 대해 커미트된 오프셋을 유지보수합니다. 이용자는 메시지를 처리할 때 파티션에서 제거하지 않습니다. 대신, 오프셋을 커미트라는 프로세스를 사용하여 현재 오프셋을 업데이트만 합니다.

Event Streams는 커미트된 오프셋 정보를 7일 동안 보존합니다.

메시지 이용 속도 제어

메시지 폭주로 인해 메시지 처리에 문제가 있는 경우, 소비자 옵션을 설정하여 메시지 소비 속도를 제어할 수 있습니다. fetch.max.bytes 및 max.poll.records를 사용하여 poll()에 대한 호출이 리턴할 수 있는 데이터 양을 제어하십시오.

이용자 재조정 처리

소비자가 그룹에 추가되거나 그룹에서 제거되면 그룹 재조정이 이루어지며, 소비자는 메시지를 소비할 수 없게 됩니다. 이로 인해 소비자 그룹에 속한 모든 소비자는 단기간 동안 서비스를 이용할 수 없게 됩니다.

"파티션이 취소됨" 콜백으로 알림을 받는 경우, 자동 커밋을 사용하지 않는 경우 수동으로 오프셋을 커밋하고 "파티션이 할당됨" 콜백을 사용하여 성공적인 재조정 알림을 받을 때까지 추가 처리를 일시 중지하려면 ConsumerRebalanceListener 사용하세요.

코드 스니펫

이러한 코드 스니펫은 관련된 개념을 설명하기 위해 상위 레벨에 있습니다. 전체 예제는 GitHub의 Event Streams 샘플을 참조하십시오.

이용자를 Event Streams에 연결하려면 서비스 인증 정보를 작성해야 합니다. 이러한 신임 정보를 가져오는 방법에 대한 정보는 Event Streams 을 참조하십시오.

이용자 코드에서 먼저 구성 특성 세트를 빌드해야 합니다. Event Streams 대한 모든 연결은 TLS 및 사용자 비밀번호 인증을 사용하여 보호되므로 최소한 이러한 속성이 필요합니다. 부트스트랩_엔디포인트, 사용자, 비밀번호를 자신의 서비스 자격 증명에 있는 비밀번호로 바꿉니다:

Properties props = new Properties();
 props.put("bootstrap.servers", BOOTSTRAP_ENDPOINTS);
 props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"USER\" password=\"PASSWORD\";");
 props.put("security.protocol", "SASL_SSL");
 props.put("sasl.mechanism", "PLAIN");
 props.put("ssl.protocol", "TLSv1.2");
 props.put("ssl.enabled.protocols", "TLSv1.2");
 props.put("ssl.endpoint.identification.algorithm", "HTTPS");

메시지를 소비하려면 다음 예제에서와 같이 키와 값에 대한 역직렬화기를 지정해야 합니다.

 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
 props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

이 디시리얼라이저는 생성자가 사용하는 시리얼라이저와 일치해야 합니다.

그런 다음 KafkaConsumer를 사용하여 메시지를 이용하십시오. 여기서 각 메시지는 ConsumerRecord로 표시됩니다. 메시지를 이용하는 가장 일반적인 방법은 그룹 ID를 설정하여 이용자 그룹에 이용자를 포함시킨 다음 토픽 목록에 대해 subscribe()를 호출하는 것입니다. 이용자에게 이용할 일부 파티션이 지정되지만, 토픽의 파티션보다 더 많은 이용자가 그룹에 있으면 이용자에게 파티션이 지정되지 않을 수도 있습니다. 그 다음, 루프에서 poll()을 호출하십시오. 처리할 일괄처리 메시지가 수신되며, 각 메시지는 ConsumerRecord로 표시됩니다.

props.put("group.id", "G1");
Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("T1"));  // T1 is the topic name
while (true) {
   ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
   for (ConsumerRecord<String, String> record : records)
     System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}

이 소비자 루프는 영원히 실행되지만 Consumer.wakeup() 호출하여 다른 스레드에서 중단하여 깔끔하게 종료할 수 있습니다.

오프셋을 수동으로 커미트하려면 먼저 enable.auto.commit 구성을 false로 설정해야 합니다. 그런 다음 Consumer.commmitSync() 또는 Consumer.commitAsync()를 사용하여 이용자의 커미트된 오프셋을 주기적으로 업데이트하십시오. 단순성을 위해 이 예제는 각 파티션의 레코드를 처리하고 마지막 오프셋을 별도로 커미트합니다.

props.put("group.id", "G1");
props.put("enable.auto.commit", "false");
Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("T1"));
try {
  while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (TopicPartition tp : records.partitions()) {
      List<ConsumerRecord<String, String>> partRecords = records.records(tp);
      long lastOffset = 0;
      for (ConsumerRecord<String, String> record : partRecords) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        lastOffset = record.offset();
      }
      // having processed all the records in the above loop, we commit the partition's offset to 1 more than the last offset
      consumer.commitSync(Collections.singletonMap(tp, new OffsetAndMetadata(lastOffset + 1)));
    }
  }
}
finally {
  consumer.close();
}

예외 처리

Kafka 클라이언트를 사용하는 강력한 애플리케이션은 예상되는 특정 상황에 대한 예외를 처리해야 합니다. 일부 메소드가 비동기적이고 Future 또는 콜백을 사용하여 결과를 전달하므로 예외가 직접 처리되지 않는 경우가 있습니다. 전체 예제를 표시하는 GitHub 의 예제 코드를 확인하십시오.

코드에서 다음 예외 목록을 처리하십시오.