Error while creating Kafka Destination

[ss025]

Enviroment

• Local
• Followed this while set up- https://docs.airbyte.io/quickstart/deploy-airbyte

Source Connector

• Postgres

Current Behavior

• Show config error while creating Kafka destination

The connection tests failed.
Verification error(s) occurred for CHECK. Errors: [$.buffer_memory: is missing but it is required, $.max_block_ms: is missing but it is required, $.socket_connection_setup_timeout_max_ms: is missing but it is required, $.linger_ms: string found, number expected, $.socket_connection_setup_timeout_ms: is missing but it is required]

Configuration on UI

Client ID - An id string to pass to the server when making requests. The purpose of this is to be able to track the source of requests beyond just ip/port by allowing a logical application name to be included in server-side request logging. (e.g. airbyte-producer) 
ab-prod-t2

ACKs * - The number of acknowledgments the producer requires the leader to have received before considering a request complete. This controls the  durability of records that are sent.  
1

Compression type * - The compression type for all data generated by the producer.  
none

Batch size * - The producer will attempt to batch records together into fewer requests whenever multiple records are being sent to the same partition.  
16384

Linger ms * - The producer groups together any records that arrive in between request transmissions into a single batched request.  
100

Max in flight requests per connection * - The maximum number of unacknowledged requests the client will send on a single connection before blocking.  
5

Buffer memory * - The total bytes of memory the producer can use to buffer records waiting to be sent to the server.  
33554432

Max request size * - The maximum size of a request in bytes.  
1048576

Retries * - Setting a value greater than zero will cause the client to resend any record whose send fails with a potentially transient error.  
2147483647

Socket connection setup timeout * - The amount of time the client will wait for the socket connection to be established.  
10000

Socket connection setup max timeout * - The maximum amount of time the client will wait for the socket connection to be established. The connection setup timeout will increase exponentially for each consecutive connection failure up to this maximum.  
30000

Max block ms * - The configuration controls how long the KafkaProducer's send(), partitionsFor(), initTransactions(), sendOffsetsToTransaction(), commitTransaction() and abortTransaction() methods will block.  
60000

Request timeout * - The configuration controls the maximum amount of time the client will wait for the response of a request. If the response is not received before the timeout elapses the client will resend the request if necessary or fail the request if retries are exhausted.  
30000

Delivery timeout * - An upper bound on the time to report success or failure after a call to 'send()' returns.  
120000

Send buffer bytes * - The size of the TCP send buffer (SO_SNDBUF) to use when sending data. If the value is -1, the OS default will be used.  
131072

Receive buffer bytes * - The size of the TCP receive buffer (SO_RCVBUF) to use when reading data. If the value is -1, the OS default will be used.  
32768

Logs

airbyte-scheduler   | 2021-08-11 08:33:52 INFO i.a.w.t.TemporalAttemptExecution(get):133 - {ActivityId=f3cab8f7-63b7-3b97-a80f-3f6e55b0f5c7, ActivityType=Run, Namespace=default, RunId=d33b9dab-f762-4f02-9776-a02e3b3432f9, TaskQueue=CHECK_CONNECTION, WorkflowId=17574ec9-6907-43ac-8fed-02578673d1ff, job_log_path=/tmp/workspace/78421389-0617-42b7-a621-d9e1285468b1/0/logs.log} - Stopping cancellation check scheduling...
airbyte-server      | 2021-08-11 08:33:52 ERROR i.a.c.l.Exceptions(swallow):84 - {workspace_app_root=/tmp/workspace/server/logs} - Swallowed error.
airbyte-server      | java.lang.NullPointerException: null
airbyte-server      | 	at io.airbyte.config.persistence.ConfigRepository.getStandardDestinationDefinition(ConfigRepository.java:131) ~[io.airbyte.airbyte-config-persistence-0.29.4-alpha.jar:?]
airbyte-server      | 	at io.airbyte.scheduler.persistence.job_tracker.JobTracker.generateDestinationDefinitionMetadata(JobTracker.java:260) ~[io.airbyte.airbyte-scheduler-persistence-0.29.4-alpha.jar:?]
airbyte-server      | 	at io.airbyte.scheduler.persistence.job_tracker.JobTracker.lambda$trackCheckConnectionDestination$1(JobTracker.java:113) ~[io.airbyte.airbyte-scheduler-persistence-0.29.4-alpha.jar:?]
airbyte-server      | 	at io.airbyte.commons.lang.Exceptions.swallow(Exceptions.java:82) ~[io.airbyte-airbyte-commons-0.29.4-alpha.jar:?]
airbyte-server      | 	at io.airbyte.scheduler.persistence.job_tracker.JobTracker.trackCheckConnectionDestination(JobTracker.java:110) ~[io.airbyte.airbyte-scheduler-persistence-0.29.4-alpha.jar:?]
airbyte-server      | 	at io.airbyte.scheduler.client.DefaultSynchronousSchedulerClient.track(DefaultSynchronousSchedulerClient.java:148) ~[io.airbyte.airbyte-scheduler-client-0.29.4-alpha.jar:?]
airbyte-server      | 	at io.airbyte.scheduler.client.DefaultSynchronousSchedulerClient.execute(DefaultSynchronousSchedulerClient.java:124) ~[io.airbyte.airbyte-scheduler-client-0.29.4-alpha.jar:?]
airbyte-server      | 	at io.airbyte.scheduler.client.DefaultSynchronousSchedulerClient.createDestinationCheckConnectionJob(DefaultSynchronousSchedulerClient.java:77) ~[io.airbyte.airbyte-scheduler-client-0.29.4-alpha.jar:?]
airbyte-server      | 	at io.airbyte.scheduler.client.SpecCachingSynchronousSchedulerClient.createDestinationCheckConnectionJob(SpecCachingSynchronousSchedulerClient.java:66) ~[io.airbyte.airbyte-scheduler-client-0.29.4-alpha.jar:?]
airbyte-server      | 	at io.airbyte.server.handlers.SchedulerHandler.checkDestinationConnectionFromDestinationCreate(SchedulerHandler.java:200) ~[io.airbyte-airbyte-server-0.29.4-alpha.jar:?]
airbyte-server      | 	at io.airbyte.server.apis.ConfigurationApi.lambda$executeDestinationCheckConnection$44(ConfigurationApi.java:472) ~[io.airbyte-airbyte-server-0.29.4-alpha.jar:?]
airbyte-server      | 	at io.airbyte.server.apis.ConfigurationApi.execute(ConfigurationApi.java:560) ~[io.airbyte-airbyte-server-0.29.4-alpha.jar:?]
airbyte-server      | 	at io.airbyte.server.apis.ConfigurationApi.executeDestinationCheckConnection(ConfigurationApi.java:472) ~[io.airbyte-airbyte-server-0.29.4-alpha.jar:?]
airbyte-server      | 	at jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:?]
airbyte-server      | 	at jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:?]
airbyte-server      | 	at jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:?]
airbyte-server      | 	at java.lang.reflect.Method.invoke(Method.java:564) ~[?:?]
airbyte-server      | 	at org.glassfish.jersey.server.model.internal.ResourceMethodInvocationHandlerFactory.lambda$static$0(ResourceMethodInvocationHandlerFactory.java:52) ~[jersey-server-2.31.jar:?]
airbyte-server      | 	at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher$1.run(AbstractJavaResourceMethodDispatcher.java:124) ~[jersey-server-2.31.jar:?]
airbyte-server      | 	at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher.invoke(AbstractJavaResourceMethodDispatcher.java:167) ~[jersey-server-2.31.jar:?]
airbyte-server      | 	at org.glassfish.jersey.server.model.internal.JavaResourceMethodDispatcherProvider$TypeOutInvoker.doDispatch(JavaResourceMethodDispatcherProvider.java:219) ~[jersey-server-2.31.jar:?]
airbyte-server      | 	at org.glassfish.jersey.server.model.internal.AbstractJavaResourceMethodDispatcher.dispatch(AbstractJavaResourceMethodDispatcher.java:79) ~[jersey-server-2.31.jar:?]
airbyte-server      | 	at org.glassfish.jersey.server.model.ResourceMethodInvoker.invoke(ResourceMethodInvoker.java:469) ~[jersey-server-2.31.jar:?]
airbyte-server      | 	at org.glassfish.jersey.server.model.ResourceMethodInvoker.apply(ResourceMethodInvoker.java:391) ~[jersey-server-2.31.jar:?]
airbyte-server      | 	at org.glassfish.jersey.server.model.ResourceMethodInvoker.apply(ResourceMethodInvoker.java:80) ~[jersey-server-2.31.jar:?]
airbyte-server      | 	at org.glassfish.jersey.server.ServerRuntime$1.run(ServerRuntime.java:253) ~[jersey-server-2.31.jar:?]
airbyte-server      | 	at org.glassfish.jersey.internal.Errors$1.call(Errors.java:248) ~[jersey-common-2.31.jar:?]
airbyte-server      | 	at org.glassfish.jersey.internal.Errors$1.call(Errors.java:244) ~[jersey-common-2.31.jar:?]
airbyte-server      | 	at org.glassfish.jersey.internal.Errors.process(Errors.java:292) ~[jersey-common-2.31.jar:?]
airbyte-server      | 	at org.glassfish.jersey.internal.Errors.process(Errors.java:274) ~[jersey-common-2.31.jar:?]
airbyte-server      | 	at org.glassfish.jersey.internal.Errors.process(Errors.java:244) ~[jersey-common-2.31.jar:?]
airbyte-server      | 	at org.glassfish.jersey.process.internal.RequestScope.runInScope(RequestScope.java:265) ~[jersey-common-2.31.jar:?]
airbyte-server      | 	at org.glassfish.jersey.server.ServerRuntime.process(ServerRuntime.java:232) ~[jersey-server-2.31.jar:?]
airbyte-server      | 	at org.glassfish.jersey.server.ApplicationHandler.handle(ApplicationHandler.java:680) ~[jersey-server-2.31.jar:?]
airbyte-server      | 	at org.glassfish.jersey.servlet.WebComponent.serviceImpl(WebComponent.java:394) ~[jersey-container-servlet-core-2.31.jar:?]
airbyte-server      | 	at org.glassfish.jersey.servlet.WebComponent.service(WebComponent.java:346) ~[jersey-container-servlet-core-2.31.jar:?]
airbyte-server      | 	at org.glassfish.jersey.servlet.ServletContainer.service(ServletContainer.java:366) ~[jersey-container-servlet-core-2.31.jar:?]
airbyte-server      | 	at org.glassfish.jersey.servlet.ServletContainer.service(ServletContainer.java:319) ~[jersey-container-servlet-core-2.31.jar:?]
airbyte-server      | 	at org.glassfish.jersey.servlet.ServletContainer.service(ServletContainer.java:205) ~[jersey-container-servlet-core-2.31.jar:?]
airbyte-server      | 	at org.eclipse.jetty.servlet.ServletHolder.handle(ServletHolder.java:763) ~[jetty-servlet-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.servlet.ServletHandler.doHandle(ServletHandler.java:569) ~[jetty-servlet-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.server.handler.ScopedHandler.nextHandle(ScopedHandler.java:233) ~[jetty-server-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.server.handler.ContextHandler.doHandle(ContextHandler.java:1377) ~[jetty-server-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:188) ~[jetty-server-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.servlet.ServletHandler.doScope(ServletHandler.java:507) ~[jetty-servlet-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.server.handler.ScopedHandler.nextScope(ScopedHandler.java:186) ~[jetty-server-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.server.handler.ContextHandler.doScope(ContextHandler.java:1292) ~[jetty-server-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.server.handler.ScopedHandler.handle(ScopedHandler.java:141) ~[jetty-server-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.server.handler.HandlerWrapper.handle(HandlerWrapper.java:127) ~[jetty-server-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.server.Server.handle(Server.java:501) ~[jetty-server-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.server.HttpChannel.lambda$handle$1(HttpChannel.java:383) ~[jetty-server-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.server.HttpChannel.dispatch(HttpChannel.java:556) [jetty-server-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.server.HttpChannel.handle(HttpChannel.java:375) [jetty-server-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.server.HttpConnection.onFillable(HttpConnection.java:273) [jetty-server-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.io.AbstractConnection$ReadCallback.succeeded(AbstractConnection.java:311) [jetty-io-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.io.FillInterest.fillable(FillInterest.java:105) [jetty-io-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.io.ChannelEndPoint$1.run(ChannelEndPoint.java:104) [jetty-io-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.runTask(EatWhatYouKill.java:336) [jetty-util-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.doProduce(EatWhatYouKill.java:313) [jetty-util-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.tryProduce(EatWhatYouKill.java:171) [jetty-util-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.util.thread.strategy.EatWhatYouKill.run(EatWhatYouKill.java:129) [jetty-util-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.util.thread.ReservedThreadExecutor$ReservedThread.run(ReservedThreadExecutor.java:375) [jetty-util-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.util.thread.QueuedThreadPool.runJob(QueuedThreadPool.java:806) [jetty-util-9.4.31.v20200723.jar:9.4.31.v20200723]
airbyte-server      | 	at org.eclipse.jetty.util.thread.QueuedThreadPool$Runner.run(QueuedThreadPool.java:938) [jetty-util-9.4.31.v20200723.jar:9.4.31.v20200723]

[nord-ine]

I got this error too when trying to create a kafka destination using the airbyte UI . I think the problem is in the form parsing the values of the properties : buffer_memory, max_block_ms,linger_ms,socket_connection_setup_timeout_ms as Strings . I just looked into the spec file in https://github.com/airbytehq/airbyte/blob/master/airbyte-integrations/connectors/destination-kafka/src/main/resources/spec.json
and noticed that these properties have a number type ,not an integer type , so maybe changing their type into Integer might fix the problem .

[sherifnada]

thanks for the feedback everyone! we'll get on this very soon.

[jarrodparkes]

+1

[sherifnada]

closed in #6040