AHeise commented on code in PR #152:
URL:
https://github.com/apache/flink-connector-kafka/pull/152#discussion_r1953999555
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSink.java:
##########
@@ -110,28 +110,35 @@ public SimpleVersionedSerializer<KafkaCommittable>
getCommittableSerializer() {
@Internal
@Override
public KafkaWriter<IN> createWriter(InitContext context) throws
IOException {
- return new KafkaWriter<IN>(
- deliveryGuarantee,
- kafkaProducerConfig,
- transactionalIdPrefix,
- context,
- recordSerializer,
- context.asSerializationSchemaInitializationContext(),
- Collections.emptyList());
+ return restoreWriter(context, Collections.emptyList());
}
@Internal
@Override
public KafkaWriter<IN> restoreWriter(
- InitContext context, Collection<KafkaWriterState> recoveredState)
throws IOException {
- return new KafkaWriter<>(
- deliveryGuarantee,
- kafkaProducerConfig,
- transactionalIdPrefix,
- context,
- recordSerializer,
- context.asSerializationSchemaInitializationContext(),
- recoveredState);
+ InitContext context, Collection<KafkaWriterState> recoveredState) {
+ KafkaWriter<IN> writer;
+ if (deliveryGuarantee == DeliveryGuarantee.EXACTLY_ONCE) {
+ writer =
+ new ExactlyOnceKafkaWriter<>(
+ deliveryGuarantee,
+ kafkaProducerConfig,
+ transactionalIdPrefix,
+ context,
+ recordSerializer,
+
context.asSerializationSchemaInitializationContext(),
+ Collections.emptyList());
+ } else {
+ writer =
+ new KafkaWriter<>(
+ deliveryGuarantee,
+ kafkaProducerConfig,
+ context,
+ recordSerializer,
+
context.asSerializationSchemaInitializationContext());
+ }
+ writer.initialize();
Review Comment:
Because of the hierarchy. In the KafkaWriter we create a non-transactional
producer. That would also be created in the EOSWriter. One possibility is to
lazily create the producer on first record but that would change the current
behavior (maybe for the better in case of recovery). Not sure if we want to mix
that or follow-up.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
