Przepływ konfiguracji konektora źródłowego
Przepływ konfiguracji konektora źródłowego
Ta strona opisuje wspólne kroki konfiguracji konektorów źródłowych GIFROST. Strony poszczególnych baz zawierają parametry specyficzne dla PostgreSQL, Oracle, MySQL, SQL Server, Db2, Informix i MongoDB.
Kroki kreatora
Basic
W tej sekcji parametrem obowiązkowym jest name. Musi być unikalny w wybranym klastrze Kafka Connect i na jego podstawie GIFROST automatycznie tworzy Debezium topic.prefix.
Konwencja generowania topic.prefix:
<connector-name>-server
Topiki tworzone przez connector mają strukturę:
<topic.prefix>.<schema>.<table>
Przykład: connector orders-pg, który replikuje tabelę public.customer, zapisuje dane do:
orders-pg-server.public.customer
tasks.max dla relacyjnych źródłowych connectorów CDC pozostaje ustawione na 1. Odczyt logów transakcyjnych jest uporządkowany i w praktyce jednowątkowy, więc zwiększenie liczby tasków nie przyspiesza czytania logu. Pola tag, config.action.reload i wymuszanie liczby tasków są parametrami operacyjnymi GIFROST.
Converters
Domyślnie GIFROST zapisuje klucze i wartości rekordów w Kafce w formacie Avro. Konwerter Avro rejestruje schematy w Schema Registry, dzięki czemu konsumenci pracują na wersjonowanych schematach.
| Parametr | Domyślnie / przykład | Opis |
|---|---|---|
key.converter | io.confluent.connect.avro.AvroConverter | Serializuje klucze rekordów Kafka jako Avro. |
value.converter | io.confluent.connect.avro.AvroConverter | Serializuje wartości rekordów Kafka jako Avro. |
key.converter.schema.registry.url | http://kafka-schema-registry:8081 | Adres Schema Registry dla schematów kluczy. |
value.converter.schema.registry.url | http://kafka-schema-registry:8081 | Adres Schema Registry dla schematów wartości. |
value.converter.ignore.default.for.nullables | true | Zachowuje rzeczywiste wartości NULL dla pól nullable zamiast zastępować je wartościami domyślnymi Avro. W CDC jest to ważne, bo NULL z bazy musi pozostać odróżnialny od defaultu. |
value.converter.enhanced.avro.schema.support | true | Zachowuje dodatkowe informacje schematu Avro, np. namespace i bogatsze metadane schematu. Poprawia kompatybilność z konsumentami zależnymi od dokładnej struktury schematu. |
Database Connection
Krok Database Connection korzysta z połączeń zapisanych w Database Store. GIFROST filtruje listę według typu konektora: connector PostgreSQL pokazuje połączenia PostgreSQL, Oracle pokazuje połączenia Oracle itd.
Po wybraniu połączenia należy kliknąć Validation, aby sprawdzić host, port, nazwę bazy, użytkownika i hasło z perspektywy wybranego klastra Kafka Connect.
Zakładka Raw pokazuje konkretne parametry wygenerowane z wpisu Database Store, np. database.hostname, database.port, database.dbname, database.user oraz pola związane z hasłem. Hasło nie powinno być widoczne jawnie.
Parametry Raw można edytować, ale nie jest to zalecane. Inne funkcje GIFROST traktują wpis w Database Store jako źródło prawdy. Jeżeli dane połączenia są błędne, należy najpierw utworzyć albo poprawić credential w Database Store, a dopiero potem użyć go w kreatorze.
Database Settings
Database Settings zawiera parametry Debezium sterujące mechaniką replikacji. W standardowym scenariuszu nie trzeba ich zmieniać.
Dla PostgreSQL GIFROST stosuje konwencje:
| Parametr | Domyślnie / konwencja | Znaczenie |
|---|---|---|
slot.name | debezium_<connector-name> | Slot replikacyjny używany do odczytu zmian z WAL. |
publication.name | dbz_publication | Publikacja PostgreSQL używana przez pgoutput. |
publication.autocreate.mode | all_tables | Debezium może automatycznie utworzyć publikację dla tabel dostępnych dla użytkownika replikacyjnego. |
slot.drop.on.stop | true / enabled | Usuwa slot przy kontrolowanym zatrzymaniu connectora. Ogranicza ryzyko zalegania WAL, ale restart nie wznowi pracy ze starej pozycji slotu. |
Database Wizard
Database Wizard definiuje zakres replikacji: schematy, tabele i kolumny. GIFROST celowo startuje z bezpiecznego stanu, w którym nic nie jest wybrane do replikacji. Wybranie All Schemas rozszerza zakres na całą widoczną bazę.
Pod spodem kreator zapisuje filtry Debezium:
schema.include.list
schema.exclude.list
table.include.list
table.exclude.list
column.include.list
column.exclude.list
Filtry są wyrażeniami regularnymi. Aby bezpiecznie replikować jedną tabelę, należy wskazać schemat, a następnie tabelę. Ręczne wpisanie samego wzorca tabeli bez zawężenia schematu może objąć więcej obiektów niż oczekiwano.
Klucze wiadomości
message.key.columns jest konfigurowane przez Columns PK Mapping. To ważne dla tabel bez klucza głównego, ponieważ Debezium potrzebuje stabilnego klucza wiadomości Kafka do poprawnego reprezentowania update i delete.
Przycisk Generate Message Key Columns generuje wpisy message.key.columns dla tabel, które wymagają własnego klucza.
Topic Creation
Topic Creation określa, jak tworzone są topiki Kafka dla replikowanych tabel.
| Parametr | Domyślnie / przykład | Opis |
|---|---|---|
topic.creation.groups | default | Lista grup tworzenia topiców. |
topic.creation.default.replication.factor | 1 | Replication factor dla tworzonych topiców. |
topic.creation.default.partitions | 1 | Liczba partycji. Dla relacyjnego CDC zostawić 1, jeśli kolejność zdarzeń ma znaczenie. |
topic.creation.default.cleanup.policy | delete | Polityka czyszczenia. compact można stosować, gdy topic ma przechowywać ostatnią wartość per klucz. |
topic.creation.default.compression.type | producer | Kompresja dziedziczona z producenta. |
topic.creation.default.delete.retention.ms | 604800000 | Siedem dni. Opóźnione czyszczenie zostawia tombstone'y i znaczniki usunięcia dostępne zanim Kafka je usunie. |
Default group dotyczy wszystkich replikowanych tabel, chyba że pasuje do nich osobna grupa. Grupy niestandardowe pozwalają np. ustawić dwa dni retencji dla dużych tabel i siedem dni dla mniejszych.
Snapshot
Snapshot decyduje, czy connector ma załadować istniejące dane przed rozpoczęciem streamingu zmian z logów.
snapshot.mode | Znaczenie |
|---|---|
initial | Domyślnie. Wykonuje snapshot przy braku offsetów, potem streamuje zmiany. |
always | Wykonuje snapshot przy każdym starcie, potem streamuje zmiany. |
initial_only | Wykonuje snapshot i zatrzymuje connector przed streamingiem zmian. |
no_data | Pomija snapshot danych i zaczyna streaming z dostępnej pozycji logu. |
never | Starszy albo connector-specyficzny wariant pomijania snapshotu; używać tylko gdy wspiera go wybrana wersja. |
when_needed | Wykonuje snapshot tylko wtedy, gdy offsetów brakuje albo nie da się ich użyć. |
configuration_based | Steruje zachowaniem przez dodatkowe parametry snapshot.mode.configuration.based.*. |
custom | Używa własnej implementacji snapshottera wskazanej przez snapshot.mode.custom.name. |
Formaty dla snapshotów filtrowanych:
snapshot.include.collection.list=<schema>.<table>,<schema>.<table>
snapshot.select.statement.overrides=<schema>.<table>
snapshot.select.statement.overrides.<schema>.<table>=SELECT * FROM <schema>.<table> WHERE <condition>
Signal, Notification i Heartbeat
Sygnały pozwalają sterować Debezium po utworzeniu connectora, np. uruchomić incremental snapshot. Tabela sygnałów jest zwykle wskazywana jako:
signal.data.collection=<schemaName>.<tableName>
Notyfikacje Debezium informują o stanie dłuższych operacji, np. postępie snapshotu. GIFROST używa domyślnej nazwy topicu:
<connector-name>-notifications
Incremental snapshot używa mechanizmu watermarkingu do oznaczania okien snapshotu. Connector czyta dane chunkami i równolegle streamuje zmiany z logów, a watermarki pomagają odróżnić granice chunków od bieżących zdarzeń CDC.
Connector
Krok Connector zawiera ustawienia runtime i obsługi danych.
| Obszar | Parametry | Cel |
|---|---|---|
| Data handling | decimal.handling.mode, interval.handling.mode, time.precision.mode, binary.handling.mode, pola JSON, unavailable.value.placeholder | Kontroluje reprezentację wartości z bazy w rekordach Kafka. |
| Performance | poll.interval.ms, max.batch.size, max.queue.size, skipped.operations | Steruje pollingiem, batchowaniem, kolejką i pomijanymi typami operacji. |
| Failure handling | event.processing.failure.handling.mode, event.deserialization.failure.handling.mode | Określa zachowanie przy błędach przetwarzania zdarzeń. |
| Connector monitoring | internal.advanced.metrics.enable, custom.metric.tags | Włącza rozszerzone metryki GIFROST i tagi dashboardowe. |
Zalecamy włączenie connector monitoring. Domyślny tag metryczny:
connectorName=<connector-name>
Dodatkowe tagi można dodać, aby wzbogacić dashboardy i metryki operacji create, update, delete oraz zdarzeń filtrowanych.
Exactly Once Support
exactly.once.support=requested prosi Kafka Connect i connector o użycie exactly-once, jeśli środowisko to wspiera. transaction.boundary=poll ustawia granicę transakcji na cykl poll, a transaction.boundary.interval.ms może ograniczyć czas trwania transakcji.
Dla pipeline'ów CDC najważniejsza pozostaje kolejność zmian, więc te ustawienia należy zmieniać tylko świadomie.
Events
Krok Events konfiguruje strukturę zdarzeń i dodatkowe przetwarzanie.
| Parametr | Cel |
|---|---|
converters | Lista customowych konwerterów wartości. |
post.processors | Post-processory zdarzeń zmian. |
topic.naming.strategy | Klasa budująca nazwy topiców, np. io.debezium.schema.SchemaTopicNamingStrategy. |
transaction.metadata.factory | Fabryka struktur metadanych transakcyjnych. |
table.ignore.builtin | Pomija tabele systemowe, jeśli connector to wspiera. |
column.mask.hash.<N>.with.salt.<M> | Maskuje kolumny wrażliwe hashem z solą. |
column.mask.with.<N>.chars | Zastępuje wartości kolumn określoną liczbą znaków maskujących. |
column.truncate.to.<N>.chars | Ucina długie wartości przed emisją. |
include.schema.comments | Dodaje komentarze tabel i kolumn do schematów. |
column.propagate.source.type | Przenosi metadane oryginalnego typu kolumny. |
Transforms
Transforms to Kafka Connect Single Message Transforms wykonywane po utworzeniu rekordu przez Debezium i przed zapisem do Kafki.
transforms=unwrap
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
predicates i predicates.config pozwalają uruchamiać transformacje warunkowo.
Error Handling
| Parametr | Cel |
|---|---|
errors.retry.timeout | Jak długo Kafka Connect ponawia operacje po błędzie. |
errors.retry.delay.max.ms | Maksymalna przerwa między retry. |
errors.tolerance | none zatrzymuje connector na błędzie; all kontynuuje i wymaga ostrożności. |
errors.log.enable | Włącza logowanie błędów. |
errors.log.include.messages | Dodaje payload rekordów do logów; pomaga w diagnostyce, ale może ujawnić dane wrażliwe. |
Summary
Ostatni krok pokazuje JSON przekazywany do Kafka Connect. Przed uruchomieniem należy sprawdzić name, wygenerowany topic.prefix, połączenie, zakres replikacji, snapshot mode, topic creation groups, transformacje i error handling.