Skip to content

Feat/main codaxy/kafka status producer #5

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

Merged
tymarats merged 5 commits into from
Feb 13, 2025
Merged

Feat/main codaxy/kafka status producer #5

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
d1d9f5c
Make KafkaProducer static to avoid instantiating it for every message.
igor-pejakovic-codaxy Feb 11, 2025
5626be9
Make the getKafkaProducer method synchronous and type it more strongly.
igor-pejakovic-codaxy Feb 11, 2025
89d7beb
apply gradlew. :conductor-kafka:spotlessApply
igor-pejakovic-codaxy Feb 12, 2025
d6a56d6
Do not use try with resource for the producer
igor-pejakovic-codaxy Feb 13, 2025
38b0fd1
Run spotlessApply, again 🙄
igor-pejakovic-codaxy Feb 13, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
77 changes: 41 additions & 36 deletions ...c/main/java/com/netflix/conductor/kafka/status/events/services/KafkaEventServiceImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,33 @@
public class KafkaEventServiceImpl implements KafkaEventService {

private static final Logger LOGGER = LoggerFactory.getLogger(KafkaEventServiceImpl.class);
private static KafkaProducer kafkaProducer;

private static synchronized <V> KafkaProducer<String, V> getKafkaProducer(
KafkaProperties kafkaProperties) throws Exception {
if (kafkaProducer == null) {
Properties producerConfig = new Properties();
producerConfig.put(
ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
producerConfig.put(ProducerConfig.ACKS_CONFIG, "all");
producerConfig.put(
ProducerConfig.CLIENT_ID_CONFIG, InetAddress.getLocalHost().getHostName());
producerConfig.put(
ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

KafkaJsonSerializer<V> jsonSerializer = new KafkaJsonSerializer<>();
try {
KafkaProducer<String, V> producer =
new KafkaProducer<>(producerConfig, new StringSerializer(), jsonSerializer);
return producer;
} catch (Exception e) {
LOGGER.error("Failed to create producer.", e);
throw e;
}
}

return kafkaProducer;
}

@Autowired private final KafkaProperties kafkaProperties;

Expand All @@ -37,42 +64,20 @@ public KafkaEventServiceImpl(KafkaProperties kafkaProperties) {

@Override
public <V> void produce(String key, V message, String topic) throws Exception {
KafkaProducer<String, V> producer = getKafkaProducer(kafkaProperties);
final ProducerRecord<String, V> record = new ProducerRecord<>(topic, key, message);

Properties producerConfig = new Properties();
producerConfig.put(
ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getBootstrapServers());
producerConfig.put(ProducerConfig.ACKS_CONFIG, "all");
producerConfig.put(
ProducerConfig.CLIENT_ID_CONFIG, InetAddress.getLocalHost().getHostName());
producerConfig.put(
ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

KafkaJsonSerializer<V> jsonSerializer = new KafkaJsonSerializer<>();

try (KafkaProducer<String, V> producer =
new KafkaProducer<>(producerConfig, new StringSerializer(), jsonSerializer)) {

final ProducerRecord<String, V> record = new ProducerRecord<>(topic, key, message);

producer.send(
record,
(metadata, exception) -> {
if (exception != null) {
LOGGER.error(
"Failed to send message to Kafka topic: {}", topic, exception);
} else {
LOGGER.info(
"Message sent to topic: {} with offset: {}",
topic,
metadata.offset());
}
});

producer.flush();

} catch (Exception e) {
LOGGER.error("Failed to create producer.", e);
throw e;
}
producer.send(
record,
(metadata, exception) -> {
if (exception != null) {
LOGGER.error("Failed to send message to Kafka topic: {}", topic, exception);
} else {
LOGGER.info(
"Message sent to topic: {} with offset: {}",
topic,
metadata.offset());
}
});
}
}
Loading