feat: kafka config

[HungLV46]

Add Kafka features

Notes

1. Add 2 plug-ins to postgres (debezium/postgres docker image)

• postgres-decoderbufs
• wal2json⁠

2. zookeeper, kafka and kafka-connect

• Create topic (run inside kafka container)

kafka-topics --create --bootstrap-server localhost:9092 --partitions 1  --replication-factor 1 --topic aura-serenity.indexer.public.nft_transfer_events
kafka-topics --create --bootstrap-server localhost:9092 --partitions 1  --replication-factor 1 --topic aura-serenity.indexer.public.fill_events_2
kafka-topics --create --bootstrap-server localhost:9092 --partitions 1  --replication-factor 1 --topic aura-serenity.indexer.public.tokens
kafka-topics --create --bootstrap-server localhost:9092 --partitions 1  --replication-factor 1 --topic aura-serenity.indexer.public.collections
kafka-topics --create --bootstrap-server localhost:9092 --partitions 1  --replication-factor 1 --topic aura-serenity.indexer.public.token_attributes
kafka-topics --create --bootstrap-server localhost:9092 --partitions 1  --replication-factor 1 --topic aura-serenity.indexer.public.orders

• Create Kafka Connector

curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '{ "name": "indexer-connector", "config": { "connector.class": "io.debezium.connector.postgresql.PostgresConnector", "tasks.max": "1", "database.hostname": "postgres", "database.port": "5432", "database.user": "postgres", "database.password": "password", "database.dbname": "postgres", "topic.prefix": "aura-serenity.indexer", "table.include.list": "public.collections,public.fill_events_2,public.nft_transfer_events,public.orders,public.token_attributes,public.tokens", "schema.history.internal.kafka.bootstrap.servers": "kafka:9092" } }'

• In case of creating wrong connector, delete it with the following API then recreate

curl -i -X DELETE localhost:8083/connectors/<connector-name>

3 - indexer .env file add:

# For kafka
DO_KAFKA_WORK=1
KAFKA_PARTITIONS_CONSUMED_CONCURRENTLY=1
KAFKA_CONSUMER_GROUP_ID=reservoir-indexer
KAFKA_BROKERS=127.0.0.1:29092
KAFKA_CLIENT_ID=reservoir-indexer
KAFKA_MAX_BYTES_PER_PARTITION=1024