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

개요

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

설정

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

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

돌아가기

시작

모든 속성