token.java

public interface TokenManager {
    enum VerificationResult {
        OK, INVALID;
    }

    byte[] generate();

    VerificationResult verify(byte[] token);

}

public class WithLuhnCheckSum implements TokenManager {
    public final length;
    public final checksum;
    public final TokenManager delegate;

    @Override
    public byte[] generate() {
        return delegate.generate() + generateCheckSum();
    }

    public byte[] generateCheckSum() {
        return 'a';
    }

    @Override
    public VerificationResult verify(byte[] token) {
        if(verify(token[0:2])) {
            return delegate.verify(token[0:-2]);
        } else {
            return EXPIRED;
        }
    }
}

public class WithSaveToKafkaManager implements TokenManager {
    public final Producer producer;
    public final TokenManager delegate;

    @Override
    public byte[] generate() {
        byte[] generate = delegate.generate();
        save(generate);
        return generate;
    }

    @Override
    public VerificationResult verify(byte[] token) {
        return delegate.verify();
    }
}

public interface TokenVerificator {
    VerificationResult verify(byte[] token, TokenMeta);
}

class KafkaAwareTokenVerificator implements TokenManager {
    TokenManager delegate;
    List<TokenVerificator> verifications;
    
    @Override
    public byte[] generate() {
        return delegate.generate();
    }

    @Override
    public VerificationResult verify(byte[] token) {
        Meta meta = retrieveMeta(token);
        return verifications.all(_(meta));
    }
}