counts.mapValues((windowed, counter) ->
{
    try {
        return buildRecord(new LoginAttackCount(
                windowed.key(),
                counter,
                windowed.window().start(),
                windowed.window().end()));
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
})
        .toStream()
        .selectKey((k, v) -> k.toString())
        .to(TOPIC, Produced.with(stringSerde, genericAvroSerde));