I’m trying to configure kafka-connector gcs → kafka topic, but after run connector my kafka topic is empty and I cant debug it.
Parquet example in gcs
❯ parquet meta ~/Downloads/tokens_parquet2_ethereum_tokens_000000000000.parquet 20:23:49
File path: /Users/alinaglumova/Downloads/tokens_parquet2_ethereum_tokens_000000000000.parquet
Created by: parquet-cpp-arrow version 13.0.0
Properties: (none)
Schema:
message schema {
required binary address (STRING);
optional binary symbol (STRING);
optional binary name (STRING);
optional binary decimals (STRING);
optional binary total_supply (STRING);
required int64 block_timestamp (TIMESTAMP(MICROS,false));
required int64 block_number;
required binary block_hash (STRING);
}
Row group 0: count: 1068 159.64 B records start: 4 total(compressed): 166.502 kB total(uncompressed):166.502 kB
--------------------------------------------------------------------------------
type encodings count avg size nulls min / max
address BINARY _ _ R 1068 47.51 B 0 "0x005c97569a24303e9ba6de6..." / "0xffffe5b9cb42b4996997c92..."
symbol BINARY _ _ R 1068 6.21 B 10 "" / "��"
name BINARY _ _ R 1068 10.27 B 10 "" / "����������"
decimals BINARY _ _ R 1068 0.47 B 65 "0" / "9"
total_supply BINARY _ _ R 1068 6.24 B 9 "0" / "9999999999999999999900000..."
block_timestamp INT64 _ _ R 1068 9.32 B 0 "2024-02-08T15:50:47.000000" / "2024-09-11T06:57:23.000000"
block_number INT64 _ _ R 1068 9.32 B 0 "19184445" / "20725728"
block_hash BINARY _ _ R 1068 70.31 B 0 "0x0002376d87ff1bbe5310679..." / "0xffae2542617a1ee9204fb27..."
config:
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
name: eth-etl-gcs-source-parquet-connector0
labels:
strimzi.io/cluster: lambda-kafka-connect
spec:
class: io.lenses.streamreactor.connect.gcp.storage.source.GCPStorageSourceConnector
tasksMax: 1
config:
connect.gcpstorage.kcql: "
insert into eth-etl-tokens-from-parquet select * from data-lambda-ethereum-etl-tokens:tokens_parquet2 BATCH=10 STOREAS `Parquet` LIMIT 100;
"
topics: "eth-etl-tokens-from-parquet"
connect.gcpstorage.gcp.auth.mode: "File"
connect.gcpstorage.gcp.file: "${env:GOOGLE_APPLICATION_CREDENTIALS}"
connect.gcpstorage.gcp.project.id: "p2p-data-lambda"
connect.gcpstorage.error.policy: "THROW"
connect.gcpstorage.http.socket.timeout: 300000
connect.gcpstorage.source.extension.includes: "parquet"
connect.gcpstorage.source.partition.search.continuous: true
connect.gcpstorage.source.partition.search.interval: 300000
connect.gcpstorage.source.partition.search.recurse.levels: 0
errors.log.enable: "true"
errors.log.include.messages: "true"
log4j.logger.io.lenses.streamreactor.connect: "DEBUG"
log4j.logger.org.apache.parquet: "DEBUG"
log4j.logger.org.apache.hadoop: "DEBUG"
log4j.logger.io.lenses.streamreactor.connect.cloud.common: "DEBUG"
log4j.logger.io.lenses.streamreactor.connect.cloud.common.formats.reader: "DEBUG"
log4j.logger.com.google.cloud: "DEBUG"
log4j.logger.com.google.auth: "DEBUG"
log4j.logger.org.apache.kafka.connect.runtime.WorkerSourceTask: "DEBUG"
log4j.logger.org.apache.kafka.connect.runtime.WorkerTask: "DEBUG"
Topic have created in kafka but I cant see some errors in logs:
2024-09-25 14:59:12,997 DEBUG releaseRequestBuffer HttpConnection@40d7f6f8::SocketChannelEndPoint@1b3fd0ff{l=/172.23.1.12:8083,r=/172.23.1.107:44180,ISHUT,fill=-,flush=-,to=0/30000}{io=1/0,kio=1,kro=1}->HttpConnection@40d7f6f8[p=HttpParser{s=CLOSED,0 of -1},g=HttpGenerator@4d4595f6{s=END}]=>HttpChannelOverHttp@5ce6883a{s=HttpChannelState@cbe8f1{s=IDLE rs=COMPLETED os=COMPLETED is=IDLE awp=false se=false i=false al=0},r=1,c=true/true,a=IDLE,uri=//lambda-kafka-connect-connect-api.streaming.svc:8083/connectors/eth-etl-gcs-source-parquet-connector0/status,age=1} (org.eclipse.jetty.server.HttpConnection) [qtp239784619-27496]
2024-09-25 14:59:13,002 DEBUG REQUEST for //lambda-kafka-connect-connect-api.streaming.svc:8083/connectors/eth-etl-gcs-source-parquet-connector0/topics on HttpChannelOverHttp@7a8fc8b2{s=HttpChannelState@5b3e5f24{s=IDLE rs=BLOCKING os=OPEN is=IDLE awp=false se=false i=true al=0},r=1,c=false/false,a=IDLE,uri=//lambda-kafka-connect-connect-api.streaming.svc:8083/connectors/eth-etl-gcs-source-parquet-connector0/topics,age=0}
GET //lambda-kafka-connect-connect-api.streaming.svc:8083/connectors/eth-etl-gcs-source-parquet-connector0/topics HTTP/1.1
2024-09-25 14:59:13,002 DEBUG HttpConnection@7f31d32c::SocketChannelEndPoint@1b040401{l=/172.23.1.12:8083,r=/172.23.1.107:44218,OPEN,fill=-,flush=-,to=0/30000}{io=1/0,kio=1,kro=1}->HttpConnection@7f31d32c[p=HttpParser{s=CONTENT,0 of -1},g=HttpGenerator@40e06ea8{s=START}]=>HttpChannelOverHttp@7a8fc8b2{s=HttpChannelState@5b3e5f24{s=IDLE rs=BLOCKING os=OPEN is=IDLE awp=false se=false i=true al=0},r=1,c=false/false,a=IDLE,uri=//lambda-kafka-connect-connect-api.streaming.svc:8083/connectors/eth-etl-gcs-source-parquet-connector0/topics,age=0} parsed true HttpParser{s=CONTENT,0 of -1} (org.eclipse.jetty.server.HttpConnection) [qtp239784619-27640]
2024-09-25 14:59:13,002 DEBUG releaseRequestBuffer HttpConnection@7f31d32c::SocketChannelEndPoint@1b040401{l=/172.23.1.12:8083,r=/172.23.1.107:44218,OPEN,fill=-,flush=-,to=0/30000}{io=1/0,kio=1,kro=1}->HttpConnection@7f31d32c[p=HttpParser{s=CONTENT,0 of -1},g=HttpGenerator@40e06ea8{s=START}]=>HttpChannelOverHttp@7a8fc8b2{s=HttpChannelState@5b3e5f24{s=IDLE rs=BLOCKING os=OPEN is=IDLE awp=false se=false i=true al=0},r=1,c=false/false,a=IDLE,uri=//lambda-kafka-connect-connect-api.streaming.svc:8083/connectors/eth-etl-gcs-source-parquet-connector0/topics,age=0} (org.eclipse.jetty.server.HttpConnection) [qtp239784619-27640]
2024-09-25 14:59:13,003 DEBUG handle //lambda-kafka-connect-connect-api.streaming.svc:8083/connectors/eth-etl-gcs-source-parquet-connector0/topics HttpChannelOverHttp@7a8fc8b2{s=HttpChannelState@5b3e5f24{s=IDLE rs=BLOCKING os=OPEN is=IDLE awp=false se=false i=true al=0},r=1,c=false/false,a=IDLE,uri=//lambda-kafka-connect-connect-api.streaming.svc:8083/connectors/eth-etl-gcs-source-parquet-connector0/topics,age=0} (org.eclipse.jetty.server.HttpChannel) [qtp239784619-27640]
2024-09-25 14:59:13,003 DEBUG action DISPATCH HttpChannelOverHttp@7a8fc8b2{s=HttpChannelState@5b3e5f24{s=HANDLING rs=BLOCKING os=OPEN is=IDLE awp=false se=false i=true al=0},r=1,c=false/false,a=HANDLING,uri=//lambda-kafka-connect-connect-api.streaming.svc:8083/connectors/eth-etl-gcs-source-parquet-connector0/topics,age=1} (org.eclipse.jetty.server.HttpChannel) [qtp239784619-27640]
2024-09-25 14:59:13,003 DEBUG REQUEST GET /connectors/eth-etl-gcs-source-parquet-connector0/topics on HttpChannelOverHttp@7a8fc8b2{s=HttpChannelState@5b3e5f24{s=HANDLING rs=BLOCKING os=OPEN is=IDLE awp=false se=false i=true al=0},r=1,c=false/false,a=HANDLING,uri=//lambda-kafka-connect-connect-api.streaming.svc:8083/connectors/eth-etl-gcs-source-parquet-connector0/topics,age=1} (org.eclipse.jetty.server.Server) [qtp239784619-27640]
2024-09-25 14:59:13,003 DEBUG scope null||/connectors/eth-etl-gcs-source-parquet-connector0/topics @ o.e.j.s.ServletContextHandler@6ada5b01{/,null,AVAILABLE} (org.eclipse.jetty.server.handler.ContextHandler) [qtp239784619-27640]
2024-09-25 14:59:13,003 DEBUG context=||/connectors/eth-etl-gcs-source-parquet-connector0/topics @ o.e.j.s.ServletContextHandler@6ada5b01{/,null,AVAILABLE} (org.eclipse.jetty.server.handler.ContextHandler) [qtp239784619-27640]
2024-09-25 14:59:13,003 DEBUG servlet ||/connectors/eth-etl-gcs-source-parquet-connector0/topics -> org.glassfish.jersey.servlet.ServletContainer-6f8f9a99==org.glassfish.jersey.servlet.ServletContainer@3b7dbc14{jsp=null,order=-1,inst=true,async=true,src=EMBEDDED:null,STARTED} (org.eclipse.jetty.servlet.ServletHandler) [qtp239784619-27640]
2024-09-25 14:59:13,003 DEBUG COMMIT for /connectors/eth-etl-gcs-source-parquet-connector0/topics on HttpChannelOverHttp@7a8fc8b2{s=HttpChannelState@5b3e5f24{s=HANDLING rs=BLOCKING os=COMMITTED is=IDLE awp=false se=false i=true al=0},r=1,c=false/false,a=HANDLING,uri=//lambda-kafka-connect-connect-api.streaming.svc:8083/connectors/eth-etl-gcs-source-parquet-connector0/topics,age=1}
2024-09-25 14:59:13,003 DEBUG flushed 196 SocketChannelEndPoint@1b040401{l=/172.23.1.12:8083,r=/172.23.1.107:44218,OPEN,fill=-,flush=W,to=0/30000}{io=1/0,kio=1,kro=1}->HttpConnection@7f31d32c[p=HttpParser{s=CONTENT,0 of -1},g=HttpGenerator@40e06ea8{s=COMPLETING}]=>HttpChannelOverHttp@7a8fc8b2{s=HttpChannelState@5b3e5f24{s=HANDLING rs=BLOCKING os=COMMITTED is=IDLE awp=false se=false i=true al=0},r=1,c=false/false,a=HANDLING,uri=//lambda-kafka-connect-connect-api.streaming.svc:8083/connectors/eth-etl-gcs-source-parquet-connector0/topics,age=1} (org.eclipse.jetty.io.ChannelEndPoint) [qtp239784619-27640]
2024-09-25 14:59:13,003 DEBUG handled=true async=false committed=true on HttpChannelOverHttp@7a8fc8b2{s=HttpChannelState@5b3e5f24{s=HANDLING rs=BLOCKING os=COMPLETED is=IDLE awp=false se=false i=true al=0},r=1,c=false/true,a=HANDLING,uri=//lambda-kafka-connect-connect-api.streaming.svc:8083/connectors/eth-etl-gcs-source-parquet-connector0/topics,age=1} (org.eclipse.jetty.server.Server) [qtp239784619-27640]
2024-09-25 14:59:13,003 DEBUG action COMPLETE HttpChannelOverHttp@7a8fc8b2{s=HttpChannelState@5b3e5f24{s=HANDLING rs=COMPLETING os=COMPLETED is=IDLE awp=false se=false i=false al=0},r=1,c=false/true,a=HANDLING,uri=//lambda-kafka-connect-connect-api.streaming.svc:8083/connectors/eth-etl-gcs-source-parquet-connector0/topics,age=1} (org.eclipse.jetty.server.HttpChannel) [qtp239784619-27640]
2024-09-25 14:59:13,003 DEBUG action TERMINATED HttpChannelOverHttp@7a8fc8b2{s=HttpChannelState@5b3e5f24{s=IDLE rs=COMPLETED os=COMPLETED is=IDLE awp=false se=false i=false al=0},r=1,c=true/true,a=IDLE,uri=//lambda-kafka-connect-connect-api.streaming.svc:8083/connectors/eth-etl-gcs-source-parquet-connector0/topics,age=1} (org.eclipse.jetty.server.HttpChannel) [qtp239784619-27640]
2024-09-25 14:59:13,003 DEBUG onCompleted for /connectors/eth-etl-gcs-source-parquet-connector0/topics written=55 (org.eclipse.jetty.server.HttpChannel) [qtp239784619-27640]
2024-09-25 14:59:13,003 INFO 172.23.1.107 - - [25/Sep/2024:14:59:13 +0000] "GET /connectors/eth-etl-gcs-source-parquet-connector0/topics HTTP/1.1" 200 55 "-" "-" 1 (org.apache.kafka.connect.runtime.rest.RestServer) [qtp239784619-27640]
2024-09-25 14:59:13,003 DEBUG HttpConnection@7f31d32c::SocketChannelEndPoint@1b040401{l=/172.23.1.12:8083,r=/172.23.1.107:44218,OPEN,fill=-,flush=-,to=0/30000}{io=1/0,kio=1,kro=1}->HttpConnection@7f31d32c[p=HttpParser{s=CONTENT,0 of -1},g=HttpGenerator@40e06ea8{s=END}]=>HttpChannelOverHttp@7a8fc8b2{s=HttpChannelState@5b3e5f24{s=IDLE rs=COMPLETED os=COMPLETED is=IDLE awp=false se=false i=false al=0},r=1,c=true/true,a=IDLE,uri=//lambda-kafka-connect-connect-api.streaming.svc:8083/connectors/eth-etl-gcs-source-parquet-connector0/topics,age=1} filled 0 HeapByteBuffer@1f4f4e4d[p=0,l=0,c=8192,r=0]={<<<>>>GET /conn...\x00\x00\x00\x00\x00\x00\x00} (org.eclipse.jetty.server.HttpConnection) [qtp239784619-27640]
2024-09-25 14:59:13,003 DEBUG HttpConnection@7f31d32c::SocketChannelEndPoint@1b040401{l=/172.23.1.12:8083,r=/172.23.1.107:44218,OPEN,fill=-,flush=-,to=0/30000}{io=1/0,kio=1,kro=1}->HttpConnection@7f31d32c[p=HttpParser{s=CONTENT,0 of -1},g=HttpGenerator@40e06ea8{s=END}]=>HttpChannelOverHttp@7a8fc8b2{s=HttpChannelState@5b3e5f24{s=IDLE rs=COMPLETED os=COMPLETED is=IDLE awp=false se=false i=false al=0},r=1,c=true/true,a=IDLE,uri=//lambda-kafka-connect-connect-api.streaming.svc:8083/connectors/eth-etl-gcs-source-parquet-connector0/topics,age=1} parse HeapByteBuffer@1f4f4e4d[p=0,l=0,c=8192,r=0]={<<<>>>GET /conn...\x00\x00\x00\x00\x00\x00\x00} {} (org.eclipse.jetty.server.HttpConnection) [qtp239784619-27640]
2024-09-25 14:59:13,004 DEBUG onContentComplete HttpChannelOverHttp@7a8fc8b2{s=HttpChannelState@5b3e5f24{s=IDLE rs=COMPLETED os=COMPLETED is=IDLE awp=false se=false i=false al=0},r=1,c=true/true,a=IDLE,uri=//lambda-kafka-connect-connect-api.streaming.svc:8083/connectors/eth-etl-gcs-source-parquet-connector0/topics,age=2} (org.eclipse.jetty.server.HttpChannel) [qtp239784619-27640]
2024-09-25 14:59:13,004 DEBUG onRequestComplete HttpChannelOverHttp@7a8fc8b2{s=HttpChannelState@5b3e5f24{s=IDLE rs=COMPLETED os=COMPLETED is=IDLE awp=false se=false i=false al=0},r=1,c=true/true,a=IDLE,uri=//lambda-kafka-connect-connect-api.streaming.svc:8083/connectors/eth-etl-gcs-source-parquet-connector0/topics,age=2} (org.eclipse.jetty.server.HttpChannel) [qtp239784619-27640]
2024-09-25 14:59:13,004 DEBUG HttpConnection@7f31d32c::SocketChannelEndPoint@1b040401{l=/172.23.1.12:8083,r=/172.23.1.107:44218,OPEN,fill=-,flush=-,to=0/30000}{io=1/0,kio=1,kro=1}->HttpConnection@7f31d32c[p=HttpParser{s=END,0 of -1},g=HttpGenerator@40e06ea8{s=END}]=>HttpChannelOverHttp@7a8fc8b2{s=HttpChannelState@5b3e5f24{s=IDLE rs=COMPLETED os=COMPLETED is=IDLE awp=false se=false i=false al=0},r=1,c=true/true,a=IDLE,uri=//lambda-kafka-connect-connect-api.streaming.svc:8083/connectors/eth-etl-gcs-source-parquet-connector0/topics,age=2} parsed false HttpParser{s=END,0 of -1} (org.eclipse.jetty.server.HttpConnection) [qtp239784619-27640]
2024-09-25 14:59:13,004 DEBUG releaseRequestBuffer HttpConnection@7f31d32c::SocketChannelEndPoint@1b040401{l=/172.23.1.12:8083,r=/172.23.1.107:44218,OPEN,fill=-,flush=-,to=0/30000}{io=1/0,kio=1,kro=1}->HttpConnection@7f31d32c[p=HttpParser{s=END,0 of -1},g=HttpGenerator@40e06ea8{s=END}]=>HttpChannelOverHttp@7a8fc8b2{s=HttpChannelState@5b3e5f24{s=IDLE rs=COMPLETED os=COMPLETED is=IDLE awp=false se=false i=false al=0},r=1,c=true/true,a=IDLE,uri=//lambda-kafka-connect-connect-api.streaming.svc:8083/connectors/eth-etl-gcs-source-parquet-connector0/topics,age=2} (org.eclipse.jetty.server.HttpConnection) [qtp239784619-27640]
Maybe someone have experience with it?