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

개요

다음 구성 설정을 사용하여 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	Type: string Description: The name of topic to use as the dead letter queue. If you specify a value, the connector writes invalid messages to the dead letter queue topic as extended JSON strings. If you leave this setting blank, the connector does not write invalid messages to any topic. 중요한: 이 속성을 활성화하려면 `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=IsHeartbeat
predicates.IsHeartbeat.type=org.apache.kafka.connect.transforms.predicates.TopicNameMatches
predicates.IsHeartbeat.pattern=__mongodb_heartbeats

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

transforms=Extract
transforms.Extract.type=org.apache.kafka.connect.transforms.ExtractField$Key
transforms.Extract.field=<the field to extract from your Apache Kafka key>
transforms.Extract.predicate=IsHeartbeat
transforms.Extract.negate=true
# apply the default key schema as the extract transformation requires a struct object
output.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 를 참조하세요.

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

돌아가기

시작

모든 속성