feat(connect): support connect openTelemetry and log for 1.6 #2961
Conversation
woshigaopp
requested review from
1sonofqiu,
Gezi-lzq and
superhx
as code owners
| @@ -0,0 +1,385 @@ | |||
| package com.automq.log.uploader.selector.kafka; | |||
There was a problem hiding this comment.
Add Apache License header
| @@ -0,0 +1,19 @@ | |||
| plugins { | |||
There was a problem hiding this comment.
Why does automq-log-uploader have independent build.gradle instead of keeping the same pattern as others modules.
| /** | ||
| * Leader election based on Kafka consumer group membership. | ||
| */ | ||
| public class KafkaLogLeaderSelectorProvider implements LogUploaderNodeSelectorProvider { |
There was a problem hiding this comment.
What's the KafkaLogLeaderSelectorProvider used for?
| int nodeId, | ||
| Map<String, String> config) { | ||
| LogUploaderNodeSelectorType type = LogUploaderNodeSelectorType.fromString(typeString); | ||
| switch (type) { |
There was a problem hiding this comment.
What are the actual application scenarios in AutoMQ of these different types?
| this(null); | ||
| } | ||
|
|
||
| public DefaultS3LogConfig(Properties overrideProps) { |
There was a problem hiding this comment.
Maybe we could directly use bucketURI as S3Stream instead of defining separated configs.
| public void run() { | ||
| while (!Thread.currentThread().isInterrupted()) { | ||
| try { | ||
| LOGGER.info("Log upload thread is running."); |
| e.printStackTrace(System.err); | ||
| Thread.sleep(1000); | ||
| } | ||
| LOGGER.info("Log upload a thread is running."); |
| while (!Thread.currentThread().isInterrupted()) { | ||
| try { | ||
| if (closed || !config.isActiveController()) { | ||
| LOGGER.info("Log cleanup thread is running."); |
There was a problem hiding this comment.
Check and remove test code.
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
|
|
||
| public class S3RollingFileAppender extends RollingFileAppender { |
There was a problem hiding this comment.
It's a little bit complex. Maybe we could keep the almost same S3RollingFileAppender and LogUploader as before.
| log4j.appender.requestAppender.layout.ConversionPattern=[%d] %p %m (%c)%n | ||
|
|
||
| log4j.appender.cleanerAppender=com.automq.shell.log.S3RollingFileAppender | ||
| log4j.appender.cleanerAppender=com.automq.log.uploader.S3RollingFileAppender |
There was a problem hiding this comment.
It's a little bit strange to have S3RollingFileAppender in a uploader package. Maybe it should in com.automq.log package
| implementation libs.reload4j | ||
| implementation libs.nettyBuffer | ||
| implementation libs.opentelemetrySdk | ||
| implementation project(':opentelemetry') |
| } | ||
| LOGGER.info("Log upload a thread is running."); | ||
| try { | ||
| String objectKey = getObjectKey(); |
There was a problem hiding this comment.
It seams the object path is same pattern for connector and broker?
2 participants
This pull request introduces the new AutoMQ Log Uploader module, which provides asynchronous S3 log upload functionality for Log4j 1.x applications. It refactors and relocates key log uploading classes, adds configuration constants, and implements a robust configuration and startup process for the uploader. The changes are grouped into module introduction/documentation, configuration and integration, and codebase refactoring.
Module introduction & documentation:
README.mdfor the newautomq-log-uploadermodule, detailing its purpose, integration steps, configuration options, and extension points for custom setups.Configuration and integration:
build.gradlefor the new module, specifying dependencies and repository settings for building and integrating the uploader.LogConfigConstants.javato centralize all configuration keys and defaults for S3 log uploading, improving maintainability and clarity.DefaultS3LogConfig.java, which loads and normalizes configuration from properties, constructs the S3 object storage URI, and sets up leader election strategies for log uploading and cleanup.Codebase refactoring and improvements:
LogUploader.javaandLogRecorder.javafrom theautomq-shellsubmodule to the newautomq-log-uploaderpackage, decoupling them from application-specific dependencies and improving error messages in validation logic. [1] [2] [3]LogUploader.java, replacing the singleton/bean pattern with explicit configuration and thread management, and removed unused async/future initialization code. [1] [2]