0ct0pu5/kafka-ui
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

wip

Browse source
This commit is contained in:
iliax 2023-05-17 00:03:26 +04:00
parent c07fd6bc4f
commit d8f11afd8b
3 changed files with 14 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/MessagesService.java
View file

@ -149,7 +149,7 @@ public class MessagesService {
}
}
public KafkaProducer<byte[], byte[]> createProducer(KafkaCluster cluster, Map<String, Object> additionalProps) {
public static KafkaProducer<byte[], byte[]> createProducer(KafkaCluster cluster, Map<String, Object> additionalProps) {
Properties properties = new Properties();
SslPropertiesUtil.addKafkaSslProperties(cluster.getOriginalProperties().getSsl(), properties);
properties.putAll(cluster.getProperties());

22
kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/audit/AuditService.java
View file

@ -35,24 +35,25 @@ public class AuditService implements Closeable {
private final Map<String, AuditWriter> auditWriters = new ConcurrentHashMap<>();
public AuditService(ClustersProperties clustersProperties,
MessagesService messagesService,
AdminClientService adminClientService,
ClustersStorage clustersStorage) {
if (clustersProperties.getClusters() != null) {
for (var clusterProps : clustersProperties.getClusters()) {
var cluster = clustersStorage.getClusterByName(clusterProps.getName()).orElseThrow();
initialize(
cluster,
adminClientService.get(cluster).block(),
messagesService
);
ReactiveAdminClient adminClient = null;
try {
adminClient = adminClientService.get(cluster).block();
} catch (Exception e) {
printAuditInitError(cluster, "Error connect to cluster", e);
continue;
}
initialize(cluster, adminClient);
}
}
}
private void initialize(KafkaCluster cluster,
ReactiveAdminClient ac,
MessagesService messagesService) {
ReactiveAdminClient ac) {
var auditProps = cluster.getOriginalProperties().getAudit();
if (auditProps == null) {
return;
@ -63,11 +64,12 @@ public class AuditService implements Closeable {
return;
}
String auditTopicName = Optional.ofNullable(auditProps.getTopic()).orElse(DEFAULT_AUDIT_TOPIC_NAME);
KafkaProducer<byte[], byte[]> producer = null;
@Nullable KafkaProducer<byte[], byte[]> producer = null;
if (topicAudit && createTopicIfNeeded(cluster, ac, auditTopicName, auditProps)) {
producer = messagesService.createProducer(cluster, Map.of());
producer = MessagesService.createProducer(cluster, Map.of());
}
auditWriters.put(cluster.getName(), new AuditWriter(auditTopicName, producer, consoleAudit));
log.info("Audit service initialized for cluster '{}'", cluster.getName());
}
/**

1
kafka-ui-api/src/main/java/com/provectus/kafka/ui/service/audit/AuditWriter.java
View file

@ -97,6 +97,7 @@ record AuditWriter(String targetTopic,
OperationResult result,
@Nullable Object params) {
//TODO: do not render null
static final JsonMapper MAPPER = new JsonMapper();
@SneakyThrows