config: name: ecommerce import: optional:configserver:http://localhost:8888 # boot 2.4부터 부트스트랩 x application: name: USER-SERVICE 이렇게 된 경우 config name 을 가져가는게 아니라 application name 으로 yaml 을 계속 찾네요.. 흠.. 자답합니다 스프링부트 2.7.11 config, bootstrap 아래 의존성으로 교체 <!-- https://mvnrepository.com/artifact/org.springframework.cloud/spring-cloud-starter-config --> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-config</artifactId> <version>3.1.0</version> </dependency> <!-- https://mvnrepository.com/artifact/org.springframework.cloud/spring-cloud-starter-bootstrap --> <dependency> <groupId>org.springframework.cloud</groupId> <artifactId>spring-cloud-starter-bootstrap</artifactId> <version>3.1.3</version> </dependency> application.yml 에 아래 코드추가 spring: config: import: classpath:/bootstrap.yml bootstrap.yml은 동일히 작성 위 과정 거치고 돌렸더니 드디어... config 가져오네요... springboot 버전과 의존 라이브러리 최신버전과 호환이 안되었던건지 이틀 간 찾아본듯 😂 spring boot 2.7.x 버전 yml 설정 이렇게 하면 됩니다. - 인프런 | 질문 & 답변 ( inflearn.com )
안녕하세요! 좋은 강의 제공해주셔서 감사합니다. mvn spring-boot: run-Dspring-boot.run .jvmArguments='-Dserver.port=9003' 다음과 같은 명령어나 실행 옵션에서 -D와 같은 옵션을 띄어쓰기 없이 사용하는 이유가 있을까요?
안녕하세요! 좋은 강의 제공해주셔서 감사합니다. mvn spring-boot: run-Dspring-boot.run .jvmArguments='-Dserver.port=9003' 다음과 같은 명령어나 실행 옵션에서 -D와 같은 옵션을 띄어쓰기 없이 사용하는 이유가 있을까요?
IntelliJ에서 ConsumerWakeup파일을 실행하고 종료 시, logger .info ( "main program starts to exit by calling wakeup" ); 이 콘솔에 출력되지 않고 아래와 같은 오류가 출력됩니다. > Task :consumers:ConsumerWakeup.main() FAILED 2 actionable tasks: 1 executed, 1 up-to-date FAILURE: Build failed with an exception. * What went wrong: Execution failed for task ':consumers:ConsumerWakeup.main()'. > Build cancelled while executing task ':consumers:ConsumerWakeup.main()' * Try: > Run with --stacktrace option to get the stack trace. > Run with --info or --debug option to get more log output. > Run with --scan to get full insights. * Get more help at https://help.gradle.org BUILD FAILED in 11s 오후 4:02:44: 실행이 완료되었습니다 ':consumers:ConsumerWakeup.main()'. 만약, 커버리지 모드로 구동을 할 경우 아래와 같이 정상적으로 로그가 출력됩니다. [Thread-2] INFO com.example.kafka.ConsumerWakeup - main program starts to exit by calling wakeup [main] ERROR com.example.kafka.ConsumerWakeup - wakeup exception has been called [main] INFO com.example.kafka.ConsumerWakeup - finally consumer is closing Class transformation time: 0.1235649s for 1928 classes or 6.408967842323652E-5s per class VirtualBox 세팅 문제로 별개의 Bare-Metal 리눅스 서버를 구성하여 사용중입니다. Java 버전, Gradle버전, implementation환경 등은 강의와 같은 상태인데 혹시 어떤게 문제일까요?
원영님 강의를 보고 현재 스프링 프레임워크 기반의 회사 솔루션에 카프카 도입을 구현하고 있습니다. 터미널 상에서 컨슈머와 프로듀서는 잘 주고 받는데 IDE로 넘어와 적용하면 에러가 발생합니다 브로커의 로그를 확인해보니 Processing automatic preferred replica leader election (kafka.controller.KafkaController) Checking need to trigger auto leader balancing (kafka.controller.KafkaController) Topics not in preferred replica for broker 0 Map() (kafka.controller.KafkaController) 이런 로그들이 찍히는데 구글링해도 이렇다 할 해결방법을 못찾고 있습니다 컨슈머 프로퍼티에 부트스트랩 서버도 localhost:9092 프로듀서 프로퍼티의 부트스트랩 서버도 동일하며 서버 프로퍼티의 애드버타이즈드 리스너도 localhost :9092로 설정해뒀습니다 혹시 제가 빠뜨린 부분이 있을까요? 도움 주시면 감사하겠습니다..
virtual box 6.1.0 버전을 사용하고 있습니다. 강의에 사용하신 버전은 제 노트북에서 오류가 뜨고 원래는 조금 더 높은 버전이었으나 가상 server01, 02 추가 자체가 되지 않아 해당 버전으로 다시 설치했었습니다. 참고로, Fasoo DRM 이 다운되어 있지 않고 CPU VT 활성화를 해야하나해서 확인해봤더니 제 노트북이 삼성인데 BIOS에서 해당 설정값 변경이 불가능하게 되있습니다. 어떻게 오류 고치면 될까요ㅜㅜ 확장팩 설치로도 오류가 안고쳐졌습니다...
안녕하세요 강사님. 커넥터, 스키마 레지스트리 관련 질문드립니다! 현재 json형식의 데이터를 s3 sink connector를 통하여 parquet 형식으로 저장하려고 합니다. json 형식의 데이터는 키 값들이 일정하진 않습니다. 예를 들어 어떤 데이터는 { "test1":"test", "test2":"test2"} 이런식이고, 어떤 데이터는 {"test1":"test1"} 이런 식입니다. 이런 경우에도 스키마 레지스트리를 활용하여 적재가 가능할까요? 없는 키 값들에 대해선 default로 null값을 스키마에 명시하면 자동으로 null처리가 되어 들어오는지 궁금합니다. 그리고 아래처럼 커넥터에 설정을 주면 자동으로 json형식의 데이터가 들어올때 커넥터에서 스키마레지스트리를 바라보고 스키마를 읽어서 parquet로 적재가 되는건지 문의드립니다. value.converter.schema.registry.url=localhost:8080 value.converter=io.confluent.connect.avro.AvroConverter format.class=io.confluent.connect.s3.format.parquet.ParquetFormat
스키마 레지스트리 강의 듣다가 궁금해서 질문드립니다. 하위 호환성에서 쓰기 스키마 V1 읽기 스키마 V2가 있다고 교재에 나와있는데요. 저기에 나와있는 스키마 즉 쓰기 스키마는 source db의 table의 스키마이고 읽기 스키마는 동기화되는 target db의 table이 맞죠? 그리고 쓰기 스키마를 기준으로 producer와 consumer가 데이터를 serialize, deserialize 하구요? V1, V2라고 되어있고 그래서 schema registry에 있는 version으로 생각했었는데 생각해보니 만약 서로(source, target) 스키마 레지스트리에서 사용하는 버전이 다르다면 그거를 커넥터 정의할 경우 정의를 해주고 해야하는데 그런게 없어서 궁금해서 여쭤봐요~~
궁금한게 있어서 질문드립니다~~ consumer를 구현할 때 데이터를 올드 데이터가 덮어쓰는 경우나 삭제 처리 때문에 날짜를 validation하는 로직이 들어가게 되는데요 데이터를 올드 데이터가 덮어쓰는 경우는 저장소가 장애가나거나 비즈니스 로직으로인해 에러가 나고 뉴 데이터는 들어가게 되었을 때 올드 데이터가 데드레터에 들어가 있고 복구된다면 덮어쓰게 됩니다. consumer -> validation store validation 통과후 consumer -> source store 위의 과정 후에 저장하게 됩니다. 그런데 카프카 커넥터를 기반으로 했을 때는 위의 과정 처리를 어떻게 할 수 있을까요? 물론 로직이 들어가게 되므로 커넥터를 사용하지 않고 구현하는게 맞는 것 같기는한데 만약 그렇게 될 경우에는 카프카 커넥터의 사용범위가 로그성 데이터나 초기 데이터를 이관할 때 정도로 사용되는 범위가 축소될 것 같아서 질문드려 봅니다! 감사합니다
좋은 강의 감사드립니다. 덕분에 많은 걸 배워서 마음이 든든해 집니다. 그런데, 하나 궁금한 점이 있어서 여쭤보고 싶습니다. 우리 개발자들은 사실 귀찮은 걸 싫어하기 때문에 기술이 나날이 발전해 가는 것이라 생각이 되는데요, 강사님이 설명주신 부분처럼 Schema나 Payload를 계속해서 수동으로 만드는 것은 개발자 입장에서는 매우 비효율적일거라 생각이 됩니다. 따라서, 이미 저 부분을 해결하기 위한 많은 부분들이 고려되었을 것으로 예상이 되는데요, 혹시 저 부분을 편하게 등록할 수 있는 다른 방법이 있을까요? 키워드라도 주시면 찾아서 공부해 나가는데 큰 도움이 될 것 같습니다. 감사합니다.
안녕하세요. m1 맥북과 m2 pro 맥미니에서 모두 설치가 되지 않습니다. VirrualBox 는 현재 download 할 수 있는 7.0.8로 설치 했고, ubuntu 는 ubuntu-20.04.6-desktop-amd64.iso 로 설치했습니다. 처음 시작하면 console 창에 "RETBleed: WARNING: spectre v2 mitigation leaves CPU vulnerable to RETBleed attacks, data leaks posssible!" 이런 에러가 나옵니다.
안녕하세요. 'Debezium Source Connector 생성하기' 수강중에 진행이 막혀 질문드립니다. config를 register_connector로 등록할 때 발생한 이슈인데 해당 이슈 : { "error_code": 400, "message": "Connector configuration is invalid and contains the following 1 error(s):\nUnable to connect: Communications link failure\n\nThe last packet sent successfully to the server was 0 milliseconds ago. The driver has not received any packets from the server.\nYou can also find the above list of errors at the endpoint `/connector-plugins/{connectorType}/config/validate`" } = 커넥터 구성이 잘못되어 MySQL 서버에 연결할 수 없다로 해석되어 다음과 같은 해결책을 시도해보았습니다. 0. http GET http://localhost:8083/connector-plugins | jq '.[].class' 조회시 io.debezium.connector.mysql.MySqlConnector 정상등록 확인 1.MySQL 서버 정상실행중 확인 2.포트 확인 > SHOW VARIABLES LIKE 'port'; >> 3306 3.네트워크 연결 상태 확인 3-1. mysql 외부접속 허용 확인 : my.cnf에서 bind항목 점검 >> 없음=기본세팅인 허용 3-2. 방화벽 규칙 확인 : Chain INPUT, FORWARD, OUTPUT : Accept , sudo iptables -A INPUT -p tcp --dport 3306 -j ACCEPT 4.SSL 문제인가? >> my.cnf에 따로 ssl-mode 설정은 없음. 5.ifconfig >> enp0s8 inet 192.168.56.101 이렇게 확인했음에도 해결되지않아 도움요청드립니다. config { "name": "mysql_cdc_oc_source_test01", "config": { "connector.class": "io.debezium.connector.mysql.MySqlConnector", "tasks.max": "1", "database.hostname": "192.168.56.101", "database.port": "3306", "database.user": "connect_dev", "database.password": "connect_dev", "database.server.id": "10000", "database.server.name": "test01", "database.include.list": "oc", "database.allowPublicKeyRetrieval": "true", "database.history.kafka.bootstrap.servers": "192.168.56.101:9092", "database.history.kafka.topic": "schema-changes.mysql.oc", "key.converter": "org.apache.kafka.connect.json.JsonConverter", "value.converter": "org.apache.kafka.connect.json.JsonConverter" } } connect 로그는 다음과 같습니다. ERROR Failed testing connection for jdbc:mysql://192.168.56.101:3306/?useInformationSchema=true&nullCatalogMeansCurrent=false&useUnicode=true&characterEncoding=UTF-8&characterSetResults=UTF-8&zeroDateTimeBehavior=CONVERT_TO_NULL&connectTimeout=30000 with user 'connect_dev' (io.debezium.connector.mysql.MySqlConnector:103) com.mysql.cj.jdbc.exceptions.CommunicationsException: Communications link failure The last packet sent successfully to the server was 0 milliseconds ago. The driver has not received any packets from the server. at com.mysql.cj.jdbc.exceptions.SQLError.createCommunicationsException(SQLError.java:174) at com.mysql.cj.jdbc.exceptions.SQLExceptionsMapping.translateException(SQLExceptionsMapping.java:64) at com.mysql.cj.jdbc.ConnectionImpl.createNewIO(ConnectionImpl.java:829) at com.mysql.cj.jdbc.ConnectionImpl.<init>(ConnectionImpl.java:449) at com.mysql.cj.jdbc.ConnectionImpl.getInstance(ConnectionImpl.java:242) at com.mysql.cj.jdbc.NonRegisteringDriver.connect(NonRegisteringDriver.java:198) at io.debezium.jdbc.JdbcConnection.lambda$patternBasedFactory$1(JdbcConnection.java:244) at io.debezium.jdbc.JdbcConnection.connection(JdbcConnection.java:888) at io.debezium.jdbc.JdbcConnection.connection(JdbcConnection.java:883) at io.debezium.jdbc.JdbcConnection.connect(JdbcConnection.java:411) at io.debezium.connector.mysql.MySqlConnector.validateConnection(MySqlConnector.java:98) at io.debezium.connector.common.RelationalBaseSourceConnector.validate(RelationalBaseSourceConnector.java:54) at org.apache.kafka.connect.runtime.AbstractHerder.validateConnectorConfig(AbstractHerder.java:459) at org.apache.kafka.connect.runtime.AbstractHerder.lambda$validateConnectorConfig$2(AbstractHerder.java:362) at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) at java.base/java.lang.Thread.run(Thread.java:829) Caused by: com.mysql.cj.exceptions.CJCommunicationsException: Communications link failure
안녕하세요, 철민님. 항상 좋은 강의 감사드립니다! 한 가지 질문 드리고 싶은 것이 있어 글을 작성합니다. 혹시 RocksDB에서 StateStore로 읽어온 데이터는 어디에 저장이 되어 있을까요? RocksDB는 로컬에 Key / Value 형식으로 데이터를 가지는 것으로 이해를 했는데, StateStore로 읽어오면 그 값은 메모리에 있을까요? StateStore의 구현체가 RocksDB도 있고, In-Memory 같은 형태도 존재하고 있는데, 이렇게 구현체가 따로 있는 것으로 추정컨데 RocksDB를 이용한 StateStore는 값을 읽어와서 메모리에 따로 저장하지 않고, 필요할 때 마다 로컬 디스크에 있는 값을 바로바로 불러오는게 아닐까 싶은데... 어떤 형식으로 동작하는 것일까요...?
텍스트가 추가될 때마다 메시지를 전송하는 Producer의 Custom 객체 직렬화 구현 - 01~02 강의에서 kafka서버에서 kafka-console-consumer --bootstrap-server localhost:9092 --topic order-serde-topic 로 데이터 조회 시 null이 조회되는 에러가 있습니다. 원인은 " 텍스트가 추가될 때마다 메시지를 전송하는 Producer의 Custom 객체 직렬화 구현 - 01" 번 강의에서 OrderSerializer 객체에서 serialize 메소드의 serializedOrder 를 null 로 초기화 한 뒤에 값을 대입하지 않고 반환하도록 하고 지나갔네요 objectMapper.writeValueAsBytes(order); 부분을 아래와 같이 수정하면 됩니다. serializedOrder = objectMapper.writeValueAsBytes(order);
항상 답변 감사합니다. SQL server 에서 JDBC source connector 연결 구성중에 질문있어 드립니다. 환경세팅은 강의와 유사하게 진행하였습니다. timestamp와 incrementing모드로 Config json을 구성하였고 MS-SQL DB는 connect_dev 계정에 om 데이터베이스의 customers 테이블을 사용하였습니다. 쿼리는 변환하여 삽입하였습니다 { "name": "mssql_jdbc_source_customers", "config": { "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector", "tasks.max": "1", "connection.url": "jdbc:sqlserver://localhost:1433;databaseName=om;trustServerCertificate=true", "connection.user": "connect_dev", "connection.password": "connect_dev", "topic.prefix": "mssql_jdbc_", "table.whitelist": "om.customers", "poll.interval.ms": 10000, "mode": "timestamp+incrementing", "incrementing.column.name": "customer_id", "timestamp.column.name": "system_upd" } } config 역시 강의와 유사하나 차이점으론 trustServerCertificate=true 옵션을 줘서 ssl 인증 문제를 회피하였습니다. 이후 질문 드리고 싶은 주요 문제로는 아래의 에러메시지 ERROR [mssql_jdbc_source_customers|task-0] WorkerSourceTask{id=mssql_jdbc_source_customers-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask:195) org.apache.kafka.connect.errors.ConnectException: Cannot make incremental queries using timestamp columns [system_upd] on [] because all of these columns nullable. at io.confluent.connect.jdbc.source.JdbcSourceTask.validateNonNullable(JdbcSourceTask.java:546) at io.confluent.connect.jdbc.source.JdbcSourceTask.start(JdbcSourceTask.java:196) at org.apache.kafka.connect.runtime.WorkerSourceTask.initializeAndStart(WorkerSourceTask.java:225) at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:186) at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:243) at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) at java.base/java.lang.Thread.run(Thread.java:829) timestamp columns [system_upd] on [] because all of these columns nullable. (null이라 문제다) 가 원인이라는건데, SELECT TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME, IS_NULLABLE FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_NAME = 'customers' AND COLUMN_NAME = 'system_upd' 해당 쿼리를 통해 IS_NULLABLE 컬럼이 NO로 되어있음(=null이 아님)을 확인하였습니다. 이 null 문제의 경우 다른 생각해 볼 접근방법이 뭐가 있을지 여쭤보고싶습니다. incrementing 모드에서는 토픽생성까지 성공하였습니다. 감사합니다.
안녕하세요. debezium oracle connector를 생성하고 topic 만들 시에 발생하는 오류에 관련하여 질문 있습니다. 우선 오류는 [2023-05-07 20:39:26,049] INFO WorkerSourceTask{id=oracle_connector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask:505) [2023-05-07 20:39:26,049] WARN Couldn't commit processed log positions with the source database due to a concurrent connector shutdown or restart (io.debezium.connector.common.BaseSourceTask:243) [2023-05-07 20:39:31,455] WARN Couldn't resolve server kafka:9092 from bootstrap.servers as DNS resolution failed for kafka (org.apache.kafka.clients.ClientUtils:75) [2023-05-07 20:39:31,457] INFO [Producer clientId=UK-dbhistory] Closing the Kafka producer with timeoutMillis = 0 ms. (org.apache.kafka.clients.producer.KafkaProducer:1189) [2023-05-07 20:39:31,457] INFO WorkerSourceTask{id=oracle_connector-0} Committing offsets (org.apache.kafka.connect.runtime.WorkerSourceTask:488) [2023-05-07 20:39:31,458] INFO WorkerSourceTask{id=oracle_connector-0} flushing 0 outstanding messages for offset commit (org.apache.kafka.connect.runtime.WorkerSourceTask:505) [2023-05-07 20:39:31,459] ERROR WorkerSourceTask{id=oracle_connector-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connect.runtime.WorkerTask:191) org.apache.kafka.common.KafkaException: Failed to construct kafka producer at org.apache.kafka.clients.producer.KafkaProducer.<init>( KafkaProducer.java:441 ) at org.apache.kafka.clients.producer.KafkaProducer.<init>( KafkaProducer.java:301 ) at io.debezium.relational.history.KafkaDatabaseHistory.start( KafkaDatabaseHistory.java:235 ) at io.debezium.relational.HistorizedRelationalDatabaseSchema.<init>( HistorizedRelationalDatabaseSchema.java:40 ) at io.debezium.connector.oracle .OracleDatabaseSchema.<init>( OracleDatabaseSchema.java:35 ) at io.debezium.connector.oracle .OracleConnectorTask.start( OracleConnectorTask.java:55 ) at io.debezium.connector.common.BaseSourceTask.start( BaseSourceTask.java:106 ) at org.apache.kafka.connect.runtime.WorkerSourceTask.execute( WorkerSourceTask.java:232 ) at org.apache.kafka.connect.runtime.WorkerTask.doRun( WorkerTask.java:189 ) at org.apache.kafka.connect.runtime.WorkerTask.run ( WorkerTask.java:239 ) at java.util.concurrent.Executors$RunnableAdapter.call( Executors.java:511 ) at java.util.concurrent.FutureTask.run ( FutureTask.java:266 ) at java.util.concurrent.ThreadPoolExecutor.runWorker( ThreadPoolExecutor.java:1149 ) at java.util.concurrent.ThreadPoolExecutor$Worker.run( ThreadPoolExecutor.java:624 ) at java.lang.Thread.run ( Thread.java:750 ) Caused by: org.apache.kafka.common.config.ConfigException: No resolvable bootstrap urls given in bootstrap.servers at org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses( ClientUtils.java:89 ) at org.apache.kafka.clients.ClientUtils.parseAndValidateAddresses( ClientUtils.java:48 ) at org.apache.kafka.clients.producer.KafkaProducer.<init>( KafkaProducer.java:415 ) ... 14 more [2023-05-07 20:39:31,461] ERROR WorkerSourceTask{id=oracle_connector-0} Task is being killed and will not recover until manually restarted (org.apache.kafka.connect.runtime.WorkerTask:192) [2023-05-07 20:39:31,461] INFO Stopping down connector (io.debezium.connector.common.BaseSourceTask:192) [2023-05-07 20:39:31,462] WARN Could not stop task (org.apache.kafka.connect.runtime.WorkerSourceTask:175) java.lang.NullPointerException at io.debezium.connector.oracle .OracleConnectorTask.doStop( OracleConnectorTask.java:129 ) at io.debezium.connector.common.BaseSourceTask.stop( BaseSourceTask.java:206 ) at io.debezium.connector.common.BaseSourceTask.stop( BaseSourceTask.java:176 ) at org.apache.kafka.connect.runtime.WorkerSourceTask.close( WorkerSourceTask.java:173 ) at org.apache.kafka.connect.runtime.WorkerTask.doClose( WorkerTask.java:168 ) at org.apache.kafka.connect.runtime.WorkerTask.doRun( WorkerTask.java:195 ) at org.apache.kafka.connect.runtime.WorkerTask.run ( WorkerTask.java:239 ) at java.util.concurrent.Executors$RunnableAdapter.call( Executors.java:511 ) at java.util.concurrent.FutureTask.run ( FutureTask.java:266 ) at java.util.concurrent.ThreadPoolExecutor.runWorker( ThreadPoolExecutor.java:1149 ) at java.util.concurrent.ThreadPoolExecutor$Worker.run( ThreadPoolExecutor.java:624 ) at java.lang.Thread.run ( Thread.java:750 ) 이렇게 발생하는데 topic은 생성되지 않고 connector만 생성되었습니다. kafka task를 삭제하지 못하고 있다는데, 어떻게 해야하나요? connector를 삭제하고 다시 생성해도 동일한 오류가 발생합니다. -oracle connector 설정 파일입니다- { "name": "oracle_connector", "config": { "connector.class": " io.debezium.connector.oracle .OracleConnector", "key.converter": "org.apache.kafka.connect.json.JsonConverter", "key.converter.schemas.registry.url": " http://0.0.0.0:8081 ", "database.hostname": "mydb", "database.port": "port", "database.user": "user", "database.password": "userpwd", "database.dbname": "dbname", "topic.prefix": "test_", "tasks.max": "1", "database.history.kafka.bootstrap.servers": "kafka:9092", "database.history.kafka.topic": "dbhistory.testdb", " database.server.name ": "servername", " database.out.server.name ": "servername" } }
