오류 처리 및 중단 후 재개 속성

개요

다음 구성 설정을 사용하여 MongoDB Kafka 소스 커넥터에서 오류가 발생할 때 작동하는 방식을 지정하고 중단된 읽기 재개와 관련된 설정을 지정할 수 있습니다.

설정

이름	설명
mongo.errors.tolerance	유형: string 설명: 커넥터가 오류가 발생할 때 메시지 처리를 계속할지 여부입니다. 커넥터가 오류가 발생할 경우 메시지 처리를 중지하고 문제를 보고하도록 하려면 이 값을 `"none"`으로 설정합니다. `"all"` 커넥터가 메시지를 계속 처리하고 발생한 오류를 무시하도록 하려면 로 설정합니다. 중요한: 이 속성은 errors.tolerance 연결 프레임워크 속성을 재정의합니다. 기본값: `"none"` 허용되는 값: `"none"` 또는 `"all"`
mongo.errors.log.enable	유형: 불리언 설명: 커넥터가 로그 파일에 오류를 보고해야 하는지 여부입니다. 커넥터에서 발생하는 모든 오류를 기록하려면 `true`로 설정합니다. 커넥터에서 허용되지 않는 오류를 기록하려면 `false`로 설정합니다. `errors.tolerance` 또는 `mongo.errors.tolerance` 설정을 사용하여 커넥터가 허용해야 하는 오류를 지정할 수 있습니다. 중요한: 이 속성은 errors.로그.enable 연결 프레임워크 속성을 재정의합니다. 기본값: `false` 허용되는 값: `true` 또는 `false`
mongo.errors.deadletterqueue.topic.name	유형: string 설명: 데드 레터 큐로 사용할 주제의 이름입니다. 값을 지정하면 커넥터는 잘못된 메시지를 데드 레터 큐 주제에 JSON 문자열로 쓰기 (write) 작업을 수행합니다. 이 설정을 비워 두면 커넥터는 어떤 주제에도 잘못된 메시지를 쓰기 (write)지 않습니다. 중요한: 이 속성을 활성화하려면 `errors.tolerance` 또는 `mongo.errors.tolerance` 설정을 `"all"`로 설정해야 합니다. 기본값: `""` 허용되는 값: 유효한 Kafka 주제 이름
offset.partition.name	유형: string 설명: 사용할 사용자 지정 오프셋 파티션 이름입니다. 이 옵션을 사용하여 기존 오프셋에 잘못된 재개 토큰이 포함된 경우 커넥터가 새 변경 스트림을 시작하도록 지시할 수 있습니다. 이 설정을 비워 두면 커넥터는 연결 세부 정보를 기반으로 기본값 파티션 이름을 사용합니다. 오프셋 파티션 명명 전략을 보려면 저장된 오프셋 재설정 가이드를 참조하세요. 기본값: `""` 허용되는 값: string. 파티션 이름 지정에 대한 자세한 내용은 Apache Kafka API 문서의 소스 레코드를 참조하세요.
heartbeat.interval.ms	유형: long 설명: 커넥터가 하트비트 메시지 전송 사이에 대기하는 시간(밀리초)입니다. 커넥터는 소스 레코드가 지정된 간격으로 게시되지 않으면 하트비트 메시지를 보냅니다. 이 메커니즘은 볼륨이 적은 네임스페이스에 대한 커넥터의 재개 가능성을 개선합니다. 하트비트 메시지에는 `postBatchResumeToken` 데이터 필드가 포함되어 있습니다. 이 필드의 값에는 커넥터가 변경 스트림에서 마지막으로 읽은 MongoDB 서버 oplog 항목이 포함되어 있습니다. 하트비트 메시지를 비활성화하려면 `0`으로 설정합니다. 자세한 내용은 Invalid Resume Token 페이지의 방지 를 참조하세요. 기본값: `0` 허용되는 값: 정수
heartbeat.topic.name	유형: string 설명: 커넥터가 하트비트 메시지를 게시해야 하는 주제의 이름입니다. 이 기능을 활성화하려면 `heartbeat.interval.ms` 설정에 양수 값을 제공해야 합니다. 기본값: `__mongodb_heartbeats` 허용되는 값: 유효한 Kafka 주제 이름

단일 메시지 변환으로 하트비트

Kafka Connect 배포에서 하트비트를 활성화하고 SMT(단일 메시지 변환) 를 지정하는 경우 SMT에서 하트비트 메시지를 제외해야 합니다. SMT는 Atlas Stream Processing 애플리케이션을 배포할 필요 없이 소스 connector를 통과하는 메시지에 대한 변환을 지정할 수 있는 Kafka의 기능입니다.

SMT에서 하트비트 메시지를 제외하려면 조건 자를 생성하여 SMT에 적용해야 합니다. 조건자는 변환을 적용하기 전에 메시지가 조건문과 일치하는지 확인할 수 있는 SMT의 기능입니다.

다음 구성은 기본 하트비트 주제로 전송된 하트비트 메시지와 일치하는 IsHeartbeat 조건자를 정의합니다.

predicates=IsHeartbeatpredicates.IsHeartbeat.type=org.apache.kafka.connect.transforms.predicates.TopicNameMatchespredicates.IsHeartbeat.pattern=__mongodb_heartbeats

다음 구성에서는 앞의 조건자를 사용하여 ExtractField 변환에서 하트비트 메시지를 제외합니다.

transforms=Extracttransforms.Extract.type=org.apache.kafka.connect.transforms.ExtractField$Keytransforms.Extract.field=<the field to extract from your Apache Kafka key>transforms.Extract.predicate=IsHeartbeattransforms.Extract.negate=true
# apply the default key schema as the extract transformation requires a struct objectoutput.format.key=schema

이전 변환에서 하트비트 메시지를 제외하지 않으면 connector가 하트비트 메시지를 처리한 후 다음 오류가 발생합니다.

ERROR WorkerSourceTask{id=mongo-source-0} Task threw an uncaught and unrecoverable exception. Task is being killed ......Only Struct objects supported for [field extraction], found: java.lang.String

SMT에 대해 자세히 학습하려면 Confluent의 Kafka Connect에서 단일 메시지 변환을 사용하는 방법 을 참조하세요.

조건자에 대해 자세히 학습하려면 Confluent에서 필터링(Apache Kafka) 을 참조하세요.

ExtractField 변환에 대해 자세히 학습하려면 Confluent의 ExtractField 를 참조하세요.

기본 키 스키마에 대해 자세히 알아보려면 기본 스키마 페이지를 참조하세요.

돌아가기

시작

모든 속성