Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Kafka cloud events #125

Merged
merged 29 commits into from
Dec 26, 2024
Merged

Kafka cloud events #125

merged 29 commits into from
Dec 26, 2024

Conversation

jekka001
Copy link

@jekka001 jekka001 commented Oct 22, 2024 •
edited by AndriiLandiak
Loading

Pull Request description

Implementation of Kafka for processing cloud events and telemetry instead of Postgres

  • Integrated Kafka for handling cloud events and telemetry, replacing Postgres.
  • Implemented setup and launching of producers and consumers for the new topics.
  • Created a new prototype for transmitting cloud events.
  • Added configurations for new Kafka topics (cloud events and cloud events ts).
  • Added a service for synchronizing the application in queueType = kafka mode: checks for unprocessed events in Postgres and, if any are found, moves them to a Kafka topic for further processing.
  • Added new services and classes to support event processing via Kafka.
  • Improved code readability and fixed bugs.
  • Moved constants to a separate class and refactored existing classes for better code structure.
  • Added a test container for black-box testing with queueType = kafka.
  • Updated test Docker configuration to run two edge instances with different queue types: in-memory and kafka.

General checklist

  • You have reviewed the guidelines document.
  • Labels that classify your pull request have been added.
  • The milestone is specified and corresponds to fix version.
  • Description references specific issue.
  • Description contains human-readable scope of changes.
  • Description contains brief notes about what needs to be added to the documentation.
  • No merge conflicts, commented blocks of code, code formatting issues.
  • Changes are backward compatible or upgrade script is provided.
  • Similar PR is opened for PE version to simplify merge. Crosslinks between PRs added. Required for internal contributors only.

Back-End feature checklist

  • Added corresponding unit and/or integration test(s). Provide written explanation in the PR description if you have failed to add tests.
  • If new dependency was added: the dependency tree is checked for conflicts.
  • If new service was added: the service is marked with corresponding @TbCoreComponent, @TbRuleEngineComponent, @TbTransportComponent, etc.
  • If new REST API was added: the RestClient.java was updated, issue for Python REST client is created.
  • If new yml property was added: make sure a description is added (above or near the property).

@AndriiLandiak AndriiLandiak requested review from AndriiLandiak and volodymyr-babak and removed request for AndriiLandiak October 22, 2024 11:31
@AndriiLandiak AndriiLandiak added the enhancement New feature or request label Oct 22, 2024
@AndriiLandiak AndriiLandiak self-requested a review October 22, 2024 11:31
@jekka001 jekka001 changed the title [WIP] Kafka cloud events Kafka cloud events Oct 22, 2024
AndriiLandiak
AndriiLandiak reviewed Oct 25, 2024
View reviewed changes
application/src/main/java/org/thingsboard/server/controller/BaseController.java Show resolved Hide resolved
dao/src/main/java/org/thingsboard/server/dao/cloud/MessageConstants.java Outdated Show resolved Hide resolved
dao/src/main/java/org/thingsboard/server/dao/cloud/DefaultEdgeSettingsService.java Outdated Show resolved Hide resolved
common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java Outdated Show resolved Hide resolved
application/src/main/java/org/thingsboard/server/service/cloud/QueueConstants.java Outdated Show resolved Hide resolved
...ication/src/main/java/org/thingsboard/server/service/cloud/PostgresUplinkMessageService.java Outdated Show resolved Hide resolved
...ication/src/main/java/org/thingsboard/server/service/cloud/PostgresUplinkMessageService.java Outdated Show resolved Hide resolved
application/src/main/java/org/thingsboard/server/service/cloud/KafkaUplinkMessageService.java Outdated Show resolved Hide resolved
application/src/main/java/org/thingsboard/server/service/cloud/BaseUplinkMessageService.java Outdated Show resolved Hide resolved
application/src/main/java/org/thingsboard/server/service/cloud/BaseUplinkMessageService.java Outdated Show resolved Hide resolved
AndriiLandiak
AndriiLandiak reviewed Oct 25, 2024
View reviewed changes
application/src/main/java/org/thingsboard/server/service/cloud/KafkaUplinkMessageService.java Outdated
protected CloudEventService cloudEventService;

public void processHandleMessages(TenantId tenantId) {
PageData<CloudEvent> cloudEvents = findCloudEvents(tenantId);
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

max_poll_records: "${TB_QUEUE_KAFKA_MAX_POLL_RECORDS:8192}"

Based on the current config, you'll find 8192 records from Kafka and try to send them all at once. I think we could face rate limits (need to double-check it), but even without it, we need to split sending by some batch. Let's discuss it with @volodymyr-babak

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In my opinion and based on the tests I conducted, the best value is 200. It is the optimal value that yields good processing results. Of course, the fastest processing occurs with 8192.

Test results: 600,000 ts:

Value 200: processed in 4 minutes 24 seconds.
Value 8192: processed in 3 minutes 33 seconds.
During the test, 10,000 ts were sent every second for 60 seconds.

@jekka001
Use Kafka for cloud events in case KAFKA queue
a442daa 
- Renamed BaseCloudEventService to PostgresCloudEventService and refactored it.
- Created KafkaCloudEventService.
- Created queues: tb_cloud_event and tb_cloud_event_ts.
- Refactored BaseUplinkMessageService.
- Created KafkaUplinkMessageService.
- Created TbCoreCloudEventProvider.

- Refactored all components that used these services.
@jekka001
Resolve Merge
efaea08
@jekka001
Added tests for Kafka events
1d51899 
- added KafkaContainerTestSuite for start tests in kafka queue type mode
- added service second service edge, kafka and zookeeper service to docker-compose
- refactored tests
@jekka001
Added Conversation Changes
8ecc452
@jekka001
Some changes after performance tests
a41df19
@jekka001
Refactoring
355bed5 
- remove cleanupEvents from CloudEventService
AndriiLandiak
AndriiLandiak reviewed Dec 4, 2024
View reviewed changes
application/src/main/java/org/thingsboard/server/service/cloud/KafkaCloudEventService.java
@Service
@Slf4j
@ConditionalOnExpression("'${queue.type:null}'=='kafka'")
public class KafkaCloudEventService implements CloudEventService {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

KafkaEdgeGrpcSession - I like the approach that is used in this class more. It simplifies the code and does not use Postgres logic for consuming, also we use QueueConsumerManager, not creating our logic of working with a consumer, consider it in the next improvement :)

jekka001 and others added 6 commits December 4, 2024 14:16
@jekka001
Merge branch 'master' into kafka-cloud-events
6ac7fce 
# Conflicts:
#	application/src/main/java/org/thingsboard/server/service/cloud/BaseUplinkMessageService.java
#	application/src/main/java/org/thingsboard/server/service/edge/rpc/EdgeGrpcSession.java
#	application/src/main/resources/tb-edge.yml
#	common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java
#	common/queue/src/main/java/org/thingsboard/server/queue/kafka/TbKafkaTopicConfigs.java
#	common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaMonolithQueueFactory.java
#	common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbCoreQueueFactory.java
#	common/queue/src/main/java/org/thingsboard/server/queue/provider/KafkaTbRuleEngineQueueFactory.java
#	msa/edge-black-box-tests/src/test/java/org/thingsboard/server/msa/AbstractContainerTest.java
#	msa/edge-black-box-tests/src/test/java/org/thingsboard/server/msa/ContainerTestSuite.java
@jekka001
Change naming
69f1530
@jekka001
Fixed Proto
6c8defe
@jekka001
Fixed And Refactoring Migration
11076b3
@volodymyr-babak
Refactoring CloudManagerService; Use prioritized check on boolean for…
a0e42a3 
… Postgres/Kafka implementations
@volodymyr-babak
Merge remote-tracking branch 'origin/rc' into kafka-cloud-events
c003b9e
@volodymyr-babak volodymyr-babak changed the base branch from master to rc December 20, 2024 14:56
@volodymyr-babak volodymyr-babak merged commit eae3229 into thingsboard:rc Dec 26, 2024
1 of 2 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@AndriiLandiak AndriiLandiak AndriiLandiak left review comments

@volodymyr-babak volodymyr-babak Awaiting requested review from volodymyr-babak

Assignees

@jekka001 jekka001

Labels
enhancement New feature or request
Projects
None yet
Milestone
No milestone
Development

Successfully merging this pull request may close these issues.
3 participants
@jekka001 @AndriiLandiak @volodymyr-babak