@ Override public void initialize ( ) { String table = Preconditions . checkNotNull ( this . state . getProp ( ForkOperatorUtils . getPropertyNameForBranch ( JdbcPublisher . JDBC_PUBLISHER_FINAL_TABLE_NAME , this . branches , this . branchId ) ) ) ; String db = Preconditions . checkNotNull ( this . state . getProp ( ForkOperatorUtils . getPropertyNameForBranch ( JdbcPublisher . JDBC_PUBLISHER_DATABASE_NAME , this . branches , this . branchId ) ) ) ; try ( Connection conn = createConnection ( ) ) { JdbcWriterCommands commands = this . jdbcWriterCommandsFactory . newInstance ( this . state , conn ) ; Map < String , JdbcType > dateColumnMapping = commands . retrieveDateColumns ( db , table ) ; LOG . info ( "Date column mapping: " + dateColumnMapping ) ; final String dateFieldsKey = ForkOperatorUtils . getPropertyNameForBranch ( AvroToJdbcEntryConverter . CONVERTER_AVRO_JDBC_DATE_FIELDS , this . branches , this . branchId ) ; for ( WorkUnit wu : this . workUnits ) { wu . setProp ( dateFieldsKey , new Gson ( ) . toJson ( dateColumnMapping ) ) ; } } catch ( SQLException e ) { throw new RuntimeException ( e ) ; } }
AvroToJdbcEntryConverter list of date columns existing in the table . As we don t want each converter making a connection against database to get the same information . Here ConverterInitializer will retrieve it and store it into WorkUnit so that AvroToJdbcEntryConverter will use it later .