merge develop dans main

src/main/java/fr/abes/LigneKbartConnect.java

@@ -14,10 +14,10 @@
 
 @org.apache.avro.specific.AvroGenerated
 public class LigneKbartConnect extends org.apache.avro.specific.SpecificRecordBase implements org.apache.avro.specific.SpecificRecord {
-  private static final long serialVersionUID = -4747621436631233106L;
+  private static final long serialVersionUID = -6107934252124705352L;
 
 
-  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"LigneKbartConnect\",\"namespace\":\"fr.abes\",\"fields\":[{\"name\":\"PUBLICATION_TITLE\",\"type\":[\"null\",\"string\"]},{\"name\":\"PRINT_IDENTIFIER\",\"type\":[\"null\",\"string\"]},{\"name\":\"ONLINE_IDENTIFIER\",\"type\":[\"null\",\"string\"]},{\"name\":\"DATE_FIRST_ISSUE_ONLINE\",\"type\":[\"null\",\"string\"]},{\"name\":\"NUM_FIRST_VOL_ONLINE\",\"type\":[\"null\",\"string\"]},{\"name\":\"NUM_FIRST_ISSUE_ONLINE\",\"type\":[\"null\",\"string\"]},{\"name\":\"DATE_LAST_ISSUE_ONLINE\",\"type\":[\"null\",\"string\"]},{\"name\":\"NUM_LAST_VOL_ONLINE\",\"type\":[\"null\",\"string\"]},{\"name\":\"NUM_LAST_ISSUE_ONLINE\",\"type\":[\"null\",\"string\"]},{\"name\":\"TITLE_URL\",\"type\":[\"null\",\"string\"]},{\"name\":\"FIRST_AUTHOR\",\"type\":[\"null\",\"string\"]},{\"name\":\"TITLE_ID\",\"type\":[\"null\",\"string\"]},{\"name\":\"EMBARGO_INFO\",\"type\":[\"null\",\"string\"]},{\"name\":\"COVERAGE_DEPTH\",\"type\":[\"null\",\"string\"]},{\"name\":\"NOTES\",\"type\":[\"null\",\"string\"]},{\"name\":\"PUBLISHER_NAME\",\"type\":[\"null\",\"string\"]},{\"name\":\"PUBLICATION_TYPE\",\"type\":[\"null\",\"string\"]},{\"name\":\"DATE_MONOGRAPH_PUBLISHED_PRINT\",\"type\":[\"null\",\"string\"]},{\"name\":\"DATE_MONOGRAPH_PUBLISHED_ONLIN\",\"type\":[\"null\",\"string\"]},{\"name\":\"MONOGRAPH_VOLUME\",\"type\":[\"null\",\"string\"]},{\"name\":\"MONOGRAPH_EDITION\",\"type\":[\"null\",\"string\"]},{\"name\":\"FIRST_EDITOR\",\"type\":[\"null\",\"string\"]},{\"name\":\"PARENT_PUBLICATION_TITLE_ID\",\"type\":[\"null\",\"string\"]},{\"name\":\"PRECEDING_PUBLICATION_TITLE_ID\",\"type\":[\"null\",\"string\"]},{\"name\":\"ACCESS_TYPE\",\"type\":[\"null\",\"string\"]},{\"name\":\"PROVIDER_PACKAGE_PACKAGE\",\"type\":\"string\"},{\"name\":\"PROVIDER_PACKAGE_DATE_P\",\"type\":{\"type\":\"int\",\"logicalType\":\"date\"}},{\"name\":\"PROVIDER_PACKAGE_IDT_PROVIDER\",\"type\":\"int\"},{\"name\":\"BEST_PPN\",\"type\":[\"null\",\"string\"]}]}");
+  public static final org.apache.avro.Schema SCHEMA$ = new org.apache.avro.Schema.Parser().parse("{\"type\":\"record\",\"name\":\"LigneKbartConnect\",\"namespace\":\"fr.abes\",\"fields\":[{\"name\":\"PUBLICATION_TITLE\",\"type\":[\"null\",\"string\"]},{\"name\":\"PRINT_IDENTIFIER\",\"type\":[\"null\",\"string\"]},{\"name\":\"ONLINE_IDENTIFIER\",\"type\":[\"null\",\"string\"]},{\"name\":\"DATE_FIRST_ISSUE_ONLINE\",\"type\":[\"null\",\"string\"]},{\"name\":\"NUM_FIRST_VOL_ONLINE\",\"type\":[\"null\",\"string\"]},{\"name\":\"NUM_FIRST_ISSUE_ONLINE\",\"type\":[\"null\",\"string\"]},{\"name\":\"DATE_LAST_ISSUE_ONLINE\",\"type\":[\"null\",\"string\"]},{\"name\":\"NUM_LAST_VOL_ONLINE\",\"type\":[\"null\",\"string\"]},{\"name\":\"NUM_LAST_ISSUE_ONLINE\",\"type\":[\"null\",\"string\"]},{\"name\":\"TITLE_URL\",\"type\":[\"null\",\"string\"]},{\"name\":\"FIRST_AUTHOR\",\"type\":[\"null\",\"string\"]},{\"name\":\"TITLE_ID\",\"type\":[\"null\",\"string\"]},{\"name\":\"EMBARGO_INFO\",\"type\":[\"null\",\"string\"]},{\"name\":\"COVERAGE_DEPTH\",\"type\":[\"null\",\"string\"]},{\"name\":\"NOTES\",\"type\":[\"null\",\"string\"]},{\"name\":\"PUBLISHER_NAME\",\"type\":[\"null\",\"string\"]},{\"name\":\"PUBLICATION_TYPE\",\"type\":[\"null\",\"string\"]},{\"name\":\"DATE_MONOGRAPH_PUBLISHED_PRINT\",\"type\":[\"null\",\"string\"]},{\"name\":\"DATE_MONOGRAPH_PUBLISHED_ONLIN\",\"type\":[\"null\",\"string\"]},{\"name\":\"MONOGRAPH_VOLUME\",\"type\":[\"null\",\"string\"]},{\"name\":\"MONOGRAPH_EDITION\",\"type\":[\"null\",\"string\"]},{\"name\":\"FIRST_EDITOR\",\"type\":[\"null\",\"string\"]},{\"name\":\"PARENT_PUBLICATION_TITLE_ID\",\"type\":[\"null\",\"string\"]},{\"name\":\"PRECEDING_PUBLICATION_TITLE_ID\",\"type\":[\"null\",\"string\"]},{\"name\":\"ACCESS_TYPE\",\"type\":[\"null\",\"string\"]},{\"name\":\"PROVIDER_PACKAGE_PACKAGE\",\"type\":\"string\"},{\"name\":\"PROVIDER_PACKAGE_DATE_P\",\"type\":{\"type\":\"int\",\"logicalType\":\"date\"}},{\"name\":\"PROVIDER_PACKAGE_IDT_PROVIDER\",\"type\":\"int\"},{\"name\":\"ID_PROVIDER_PACKAGE\",\"type\":\"int\"},{\"name\":\"BEST_PPN\",\"type\":[\"null\",\"string\"]}]}");
   public static org.apache.avro.Schema getClassSchema() { return SCHEMA$; }
 
   private static final SpecificData MODEL$ = new SpecificData();
@@ -104,6 +104,7 @@ public static LigneKbartConnect fromByteBuffer(
   private java.lang.CharSequence PROVIDER_PACKAGE_PACKAGE;
   private java.time.LocalDate PROVIDER_PACKAGE_DATE_P;
   private int PROVIDER_PACKAGE_IDT_PROVIDER;
+  private int ID_PROVIDER_PACKAGE;
   private java.lang.CharSequence BEST_PPN;
 
   /**
@@ -143,9 +144,10 @@ public LigneKbartConnect() {}
    * @param PROVIDER_PACKAGE_PACKAGE The new value for PROVIDER_PACKAGE_PACKAGE
    * @param PROVIDER_PACKAGE_DATE_P The new value for PROVIDER_PACKAGE_DATE_P
    * @param PROVIDER_PACKAGE_IDT_PROVIDER The new value for PROVIDER_PACKAGE_IDT_PROVIDER
+   * @param ID_PROVIDER_PACKAGE The new value for ID_PROVIDER_PACKAGE
    * @param BEST_PPN The new value for BEST_PPN
    */
-  public LigneKbartConnect(java.lang.CharSequence PUBLICATION_TITLE, java.lang.CharSequence PRINT_IDENTIFIER, java.lang.CharSequence ONLINE_IDENTIFIER, java.lang.CharSequence DATE_FIRST_ISSUE_ONLINE, java.lang.CharSequence NUM_FIRST_VOL_ONLINE, java.lang.CharSequence NUM_FIRST_ISSUE_ONLINE, java.lang.CharSequence DATE_LAST_ISSUE_ONLINE, java.lang.CharSequence NUM_LAST_VOL_ONLINE, java.lang.CharSequence NUM_LAST_ISSUE_ONLINE, java.lang.CharSequence TITLE_URL, java.lang.CharSequence FIRST_AUTHOR, java.lang.CharSequence TITLE_ID, java.lang.CharSequence EMBARGO_INFO, java.lang.CharSequence COVERAGE_DEPTH, java.lang.CharSequence NOTES, java.lang.CharSequence PUBLISHER_NAME, java.lang.CharSequence PUBLICATION_TYPE, java.lang.CharSequence DATE_MONOGRAPH_PUBLISHED_PRINT, java.lang.CharSequence DATE_MONOGRAPH_PUBLISHED_ONLIN, java.lang.CharSequence MONOGRAPH_VOLUME, java.lang.CharSequence MONOGRAPH_EDITION, java.lang.CharSequence FIRST_EDITOR, java.lang.CharSequence PARENT_PUBLICATION_TITLE_ID, java.lang.CharSequence PRECEDING_PUBLICATION_TITLE_ID, java.lang.CharSequence ACCESS_TYPE, java.lang.CharSequence PROVIDER_PACKAGE_PACKAGE, java.time.LocalDate PROVIDER_PACKAGE_DATE_P, java.lang.Integer PROVIDER_PACKAGE_IDT_PROVIDER, java.lang.CharSequence BEST_PPN) {
+  public LigneKbartConnect(java.lang.CharSequence PUBLICATION_TITLE, java.lang.CharSequence PRINT_IDENTIFIER, java.lang.CharSequence ONLINE_IDENTIFIER, java.lang.CharSequence DATE_FIRST_ISSUE_ONLINE, java.lang.CharSequence NUM_FIRST_VOL_ONLINE, java.lang.CharSequence NUM_FIRST_ISSUE_ONLINE, java.lang.CharSequence DATE_LAST_ISSUE_ONLINE, java.lang.CharSequence NUM_LAST_VOL_ONLINE, java.lang.CharSequence NUM_LAST_ISSUE_ONLINE, java.lang.CharSequence TITLE_URL, java.lang.CharSequence FIRST_AUTHOR, java.lang.CharSequence TITLE_ID, java.lang.CharSequence EMBARGO_INFO, java.lang.CharSequence COVERAGE_DEPTH, java.lang.CharSequence NOTES, java.lang.CharSequence PUBLISHER_NAME, java.lang.CharSequence PUBLICATION_TYPE, java.lang.CharSequence DATE_MONOGRAPH_PUBLISHED_PRINT, java.lang.CharSequence DATE_MONOGRAPH_PUBLISHED_ONLIN, java.lang.CharSequence MONOGRAPH_VOLUME, java.lang.CharSequence MONOGRAPH_EDITION, java.lang.CharSequence FIRST_EDITOR, java.lang.CharSequence PARENT_PUBLICATION_TITLE_ID, java.lang.CharSequence PRECEDING_PUBLICATION_TITLE_ID, java.lang.CharSequence ACCESS_TYPE, java.lang.CharSequence PROVIDER_PACKAGE_PACKAGE, java.time.LocalDate PROVIDER_PACKAGE_DATE_P, java.lang.Integer PROVIDER_PACKAGE_IDT_PROVIDER, java.lang.Integer ID_PROVIDER_PACKAGE, java.lang.CharSequence BEST_PPN) {
     this.PUBLICATION_TITLE = PUBLICATION_TITLE;
     this.PRINT_IDENTIFIER = PRINT_IDENTIFIER;
     this.ONLINE_IDENTIFIER = ONLINE_IDENTIFIER;
@@ -174,6 +176,7 @@ public LigneKbartConnect(java.lang.CharSequence PUBLICATION_TITLE, java.lang.Cha
     this.PROVIDER_PACKAGE_PACKAGE = PROVIDER_PACKAGE_PACKAGE;
     this.PROVIDER_PACKAGE_DATE_P = PROVIDER_PACKAGE_DATE_P;
     this.PROVIDER_PACKAGE_IDT_PROVIDER = PROVIDER_PACKAGE_IDT_PROVIDER;
+    this.ID_PROVIDER_PACKAGE = ID_PROVIDER_PACKAGE;
     this.BEST_PPN = BEST_PPN;
   }
 
@@ -215,7 +218,8 @@ public java.lang.Object get(int field$) {
     case 25: return PROVIDER_PACKAGE_PACKAGE;
     case 26: return PROVIDER_PACKAGE_DATE_P;
     case 27: return PROVIDER_PACKAGE_IDT_PROVIDER;
-    case 28: return BEST_PPN;
+    case 28: return ID_PROVIDER_PACKAGE;
+    case 29: return BEST_PPN;
     default: throw new IndexOutOfBoundsException("Invalid index: " + field$);
     }
   }
@@ -251,6 +255,7 @@ public java.lang.Object get(int field$) {
       new org.apache.avro.data.TimeConversions.DateConversion(),
       null,
       null,
+      null,
       null
   };
 
@@ -292,7 +297,8 @@ public void put(int field$, java.lang.Object value$) {
     case 25: PROVIDER_PACKAGE_PACKAGE = (java.lang.CharSequence)value$; break;
     case 26: PROVIDER_PACKAGE_DATE_P = (java.time.LocalDate)value$; break;
     case 27: PROVIDER_PACKAGE_IDT_PROVIDER = (java.lang.Integer)value$; break;
-    case 28: BEST_PPN = (java.lang.CharSequence)value$; break;
+    case 28: ID_PROVIDER_PACKAGE = (java.lang.Integer)value$; break;
+    case 29: BEST_PPN = (java.lang.CharSequence)value$; break;
     default: throw new IndexOutOfBoundsException("Invalid index: " + field$);
     }
   }
@@ -773,6 +779,23 @@ public void setPROVIDERPACKAGEIDTPROVIDER(int value) {
     this.PROVIDER_PACKAGE_IDT_PROVIDER = value;
   }
 
+  /**
+   * Gets the value of the 'ID_PROVIDER_PACKAGE' field.
+   * @return The value of the 'ID_PROVIDER_PACKAGE' field.
+   */
+  public int getIDPROVIDERPACKAGE() {
+    return ID_PROVIDER_PACKAGE;
+  }
+
+
+  /**
+   * Sets the value of the 'ID_PROVIDER_PACKAGE' field.
+   * @param value the value to set.
+   */
+  public void setIDPROVIDERPACKAGE(int value) {
+    this.ID_PROVIDER_PACKAGE = value;
+  }
+
   /**
    * Gets the value of the 'BEST_PPN' field.
    * @return The value of the 'BEST_PPN' field.
@@ -859,6 +882,7 @@ public static class Builder extends org.apache.avro.specific.SpecificRecordBuild
     private java.lang.CharSequence PROVIDER_PACKAGE_PACKAGE;
     private java.time.LocalDate PROVIDER_PACKAGE_DATE_P;
     private int PROVIDER_PACKAGE_IDT_PROVIDER;
+    private int ID_PROVIDER_PACKAGE;
     private java.lang.CharSequence BEST_PPN;
 
     /** Creates a new Builder */
@@ -984,10 +1008,14 @@ private Builder(fr.abes.LigneKbartConnect.Builder other) {
         this.PROVIDER_PACKAGE_IDT_PROVIDER = data().deepCopy(fields()[27].schema(), other.PROVIDER_PACKAGE_IDT_PROVIDER);
         fieldSetFlags()[27] = other.fieldSetFlags()[27];
       }
-      if (isValidValue(fields()[28], other.BEST_PPN)) {
-        this.BEST_PPN = data().deepCopy(fields()[28].schema(), other.BEST_PPN);
+      if (isValidValue(fields()[28], other.ID_PROVIDER_PACKAGE)) {
+        this.ID_PROVIDER_PACKAGE = data().deepCopy(fields()[28].schema(), other.ID_PROVIDER_PACKAGE);
         fieldSetFlags()[28] = other.fieldSetFlags()[28];
       }
+      if (isValidValue(fields()[29], other.BEST_PPN)) {
+        this.BEST_PPN = data().deepCopy(fields()[29].schema(), other.BEST_PPN);
+        fieldSetFlags()[29] = other.fieldSetFlags()[29];
+      }
     }
 
     /**
@@ -1108,10 +1136,14 @@ private Builder(fr.abes.LigneKbartConnect other) {
         this.PROVIDER_PACKAGE_IDT_PROVIDER = data().deepCopy(fields()[27].schema(), other.PROVIDER_PACKAGE_IDT_PROVIDER);
         fieldSetFlags()[27] = true;
       }
-      if (isValidValue(fields()[28], other.BEST_PPN)) {
-        this.BEST_PPN = data().deepCopy(fields()[28].schema(), other.BEST_PPN);
+      if (isValidValue(fields()[28], other.ID_PROVIDER_PACKAGE)) {
+        this.ID_PROVIDER_PACKAGE = data().deepCopy(fields()[28].schema(), other.ID_PROVIDER_PACKAGE);
         fieldSetFlags()[28] = true;
       }
+      if (isValidValue(fields()[29], other.BEST_PPN)) {
+        this.BEST_PPN = data().deepCopy(fields()[29].schema(), other.BEST_PPN);
+        fieldSetFlags()[29] = true;
+      }
     }
 
     /**
@@ -2232,6 +2264,45 @@ public fr.abes.LigneKbartConnect.Builder clearPROVIDERPACKAGEIDTPROVIDER() {
       return this;
     }
 
+    /**
+      * Gets the value of the 'ID_PROVIDER_PACKAGE' field.
+      * @return The value.
+      */
+    public int getIDPROVIDERPACKAGE() {
+      return ID_PROVIDER_PACKAGE;
+    }
+
+
+    /**
+      * Sets the value of the 'ID_PROVIDER_PACKAGE' field.
+      * @param value The value of 'ID_PROVIDER_PACKAGE'.
+      * @return This builder.
+      */
+    public fr.abes.LigneKbartConnect.Builder setIDPROVIDERPACKAGE(int value) {
+      validate(fields()[28], value);
+      this.ID_PROVIDER_PACKAGE = value;
+      fieldSetFlags()[28] = true;
+      return this;
+    }
+
+    /**
+      * Checks whether the 'ID_PROVIDER_PACKAGE' field has been set.
+      * @return True if the 'ID_PROVIDER_PACKAGE' field has been set, false otherwise.
+      */
+    public boolean hasIDPROVIDERPACKAGE() {
+      return fieldSetFlags()[28];
+    }
+
+
+    /**
+      * Clears the value of the 'ID_PROVIDER_PACKAGE' field.
+      * @return This builder.
+      */
+    public fr.abes.LigneKbartConnect.Builder clearIDPROVIDERPACKAGE() {
+      fieldSetFlags()[28] = false;
+      return this;
+    }
+
     /**
       * Gets the value of the 'BEST_PPN' field.
       * @return The value.
@@ -2247,9 +2318,9 @@ public java.lang.CharSequence getBESTPPN() {
       * @return This builder.
       */
     public fr.abes.LigneKbartConnect.Builder setBESTPPN(java.lang.CharSequence value) {
-      validate(fields()[28], value);
+      validate(fields()[29], value);
       this.BEST_PPN = value;
-      fieldSetFlags()[28] = true;
+      fieldSetFlags()[29] = true;
       return this;
     }
 
@@ -2258,7 +2329,7 @@ public fr.abes.LigneKbartConnect.Builder setBESTPPN(java.lang.CharSequence value
       * @return True if the 'BEST_PPN' field has been set, false otherwise.
       */
     public boolean hasBESTPPN() {
-      return fieldSetFlags()[28];
+      return fieldSetFlags()[29];
     }
 
 
@@ -2268,7 +2339,7 @@ public boolean hasBESTPPN() {
       */
     public fr.abes.LigneKbartConnect.Builder clearBESTPPN() {
       BEST_PPN = null;
-      fieldSetFlags()[28] = false;
+      fieldSetFlags()[29] = false;
       return this;
     }
 
@@ -2305,7 +2376,8 @@ public LigneKbartConnect build() {
         record.PROVIDER_PACKAGE_PACKAGE = fieldSetFlags()[25] ? this.PROVIDER_PACKAGE_PACKAGE : (java.lang.CharSequence) defaultValue(fields()[25]);
         record.PROVIDER_PACKAGE_DATE_P = fieldSetFlags()[26] ? this.PROVIDER_PACKAGE_DATE_P : (java.time.LocalDate) defaultValue(fields()[26]);
         record.PROVIDER_PACKAGE_IDT_PROVIDER = fieldSetFlags()[27] ? this.PROVIDER_PACKAGE_IDT_PROVIDER : (java.lang.Integer) defaultValue(fields()[27]);
-        record.BEST_PPN = fieldSetFlags()[28] ? this.BEST_PPN : (java.lang.CharSequence) defaultValue(fields()[28]);
+        record.ID_PROVIDER_PACKAGE = fieldSetFlags()[28] ? this.ID_PROVIDER_PACKAGE : (java.lang.Integer) defaultValue(fields()[28]);
+        record.BEST_PPN = fieldSetFlags()[29] ? this.BEST_PPN : (java.lang.CharSequence) defaultValue(fields()[29]);
         return record;
       } catch (org.apache.avro.AvroMissingFieldException e) {
         throw e;

src/main/java/fr/abes/bestppn/configuration/KafkaConfig.java

@@ -5,7 +5,6 @@
 import io.confluent.kafka.serializers.KafkaAvroSerializer;
 import io.confluent.kafka.serializers.KafkaAvroSerializerConfig;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
-import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.apache.kafka.common.serialization.StringSerializer;
@@ -60,7 +59,7 @@ public ConsumerFactory<String, String> consumerKbartFactory() {
     }
 
     @Bean
-    public Map<String, Object> producerConfigs() {
+    public Map<String, Object> producerConfigsWithTransaction() {
         Map<String, Object> props = new HashMap<>();
         props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
         props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
@@ -72,22 +71,52 @@ public Map<String, Object> producerConfigs() {
     }
 
     @Bean
-    public ProducerFactory<String, LigneKbartConnect> producerFactory() {
-        DefaultKafkaProducerFactory<String, LigneKbartConnect> factory = new DefaultKafkaProducerFactory<>(producerConfigs());
-        factory.setTransactionIdPrefix(transactionIdPrefix);
+    public Map<String, Object> producerConfigs() {
+        Map<String, Object> props = new HashMap<>();
+        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
+        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
+        props.put(KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, registryUrl);
+        props.put(KafkaAvroSerializerConfig.AUTO_REGISTER_SCHEMAS, autoRegisterSchema);
+        return props;
+    }
+
+    @Bean
+    public ProducerFactory<String, LigneKbartConnect> producerFactoryLigneKbartConnectWithTransaction() {
+        DefaultKafkaProducerFactory<String, LigneKbartConnect> factory = new DefaultKafkaProducerFactory<>(producerConfigsWithTransaction());
+        factory.setTransactionIdPrefix(transactionIdPrefix+"connect-");
+        return factory;
+    }
+    @Bean
+    public ProducerFactory<String, LigneKbartImprime> producerFactoryLigneKbartImprimeWithTransaction() {
+        DefaultKafkaProducerFactory<String, LigneKbartImprime> factory = new DefaultKafkaProducerFactory<>(producerConfigsWithTransaction());
+        factory.setTransactionIdPrefix(transactionIdPrefix+"print-");
         return factory;
     }
 
     @Bean
-    public KafkaTransactionManager<String, LigneKbartConnect> kafkaTransactionManager(){
-        return new KafkaTransactionManager<>(producerFactory());
+    public ProducerFactory<String, String> producerFactory() {
+        DefaultKafkaProducerFactory<String, String> factory = new DefaultKafkaProducerFactory<>(producerConfigs());
+        return factory;
     }
 
+    @Bean
+    public KafkaTransactionManager<String, LigneKbartConnect> kafkaTransactionManagerKbartConnect(){
+        return new KafkaTransactionManager<>(producerFactoryLigneKbartConnectWithTransaction());
+    }
+    @Bean
+    public KafkaTransactionManager<String, LigneKbartImprime> kafkaTransactionManagerKbartImprime(){
+        return new KafkaTransactionManager<>(producerFactoryLigneKbartImprimeWithTransaction());
+    }
 
     @Bean
-    public KafkaTemplate<String, LigneKbartConnect> kafkaTemplateConnect(final ProducerFactory producerFactory) { return new KafkaTemplate<>(producerFactory);}
+    public KafkaTemplate<String, LigneKbartConnect> kafkaTemplateConnect(final ProducerFactory producerFactoryLigneKbartConnectWithTransaction) { return new KafkaTemplate<>(producerFactoryLigneKbartConnectWithTransaction);}
 
     @Bean
-    public KafkaTemplate<String, LigneKbartImprime> kafkaTemplateImprime(final ProducerFactory producerFactory) { return new KafkaTemplate<>(producerFactory);}
+    public KafkaTemplate<String, LigneKbartImprime> kafkaTemplateImprime(final ProducerFactory producerFactoryLigneKbartImprimeWithTransaction) { return new KafkaTemplate<>(producerFactoryLigneKbartImprimeWithTransaction);}
 
+    @Bean
+    public KafkaTemplate<String, String> kafkatemplateEndoftraitement(final ProducerFactory producerFactory) {
+        return new KafkaTemplate<>(producerFactory);
+    }
 }

src/main/java/fr/abes/bestppn/dto/kafka/LigneKbartDto.java

@@ -123,6 +123,8 @@ public class LigneKbartDto {
     private Date providerPackageDateP;
     @JsonProperty("provider_package_idt_provider")
     private Integer providerPackageIdtProvider;
+    @JsonProperty("id_provider_package")
+    private Integer idProviderPackage;
 
 
     @JsonIgnore