@ Override protected void write ( List < Message > msgList ) throws IOException { for ( Message msg : msgList ) { writer . writeTo ( msg ) ; String routingKey = normalizeRoutingKey ( msg ) ; DynamicCounter . increment ( MonitorConfig . builder ( "writtenMessages" ) . withTag ( TagKey . DATA_SOURCE , routingKey ) . build ( ) ) ; ++ writtenMessages ; DynamicCounter . increment ( MonitorConfig . builder ( "writtenBytes" ) . withTag ( TagKey . DATA_SOURCE , routingKey ) . build ( ) , msg . getPayload ( ) . length ) ; writtenBytes += msg . getPayload ( ) . length ; messageWrittenInRotation = true ; } writer . sync ( ) ; throughput . increment ( msgList . size ( ) ) ; }
Write all messages in msgList to file writer sync the file commit the queue and clear messages