can i print some log when a message has done?

[dragontree101]

i want to log some handle message info (time and offse and so on) when message has done,
but i only found ErrorHandler interface when consume message has error.

i don't whether other interface like ErrorHandler it can let me record message when consume has done in normal status, i will record some info?

i have read some source code , but not found

i found have commit log, when debug level is debug, but i think in pro env, log level info is the best, so i don't want to set debug level to debug.

thanks!

[garyrussell]

We could make that log level configurable (contributions are welcome - see contributing.

In the meantime, you could do something like this...

@SpringBootApplication
public class Kgh537Application {

	private static final Logger logger = LoggerFactory.getLogger(Kgh537Application.class);

	public static void main(String[] args) {
		SpringApplication.run(Kgh537Application.class, args);
	}

	@Bean
	public ApplicationRunner runner(KafkaTemplate<String, String> kafkaTemplate) {
		return args -> kafkaTemplate.send("kgh537", "foo");
	}

	@Bean
	public ConsumerFactory<Object, Object> consumerFactory(KafkaProperties properties) {
		return new DefaultKafkaConsumerFactory<Object, Object>(properties.buildConsumerProperties()) {

			@Override
			public Consumer<Object, Object> createConsumer() {
				return wrap(super.createConsumer());
			}

			@Override
			public Consumer<Object, Object> createConsumer(String groupId, String clientIdPrefix,
					String clientIdSuffix) {
				return wrap(super.createConsumer(groupId, clientIdPrefix, clientIdSuffix));
			}

			@Override
			public Consumer<Object, Object> createConsumer(String groupId, String clientIdSuffix) {
				return wrap(super.createConsumer(groupId, clientIdSuffix));
			}

			@Override
			public Consumer<Object, Object> createConsumer(String clientIdSuffix) {
				return wrap(super.createConsumer(clientIdSuffix));
			}

			@SuppressWarnings("unchecked")
			private Consumer<Object, Object> wrap(Consumer<Object, Object> createConsumer) {
				ProxyFactory pf = new ProxyFactory(createConsumer);
				pf.addAdvice(new MethodInterceptor() {

					@Override
					public Object invoke(MethodInvocation invocation) throws Throwable {
						Method method = invocation.getMethod();
						if (method.getName().startsWith("commit") && method.getReturnType().equals(void.class)) {
							invocation.proceed();
							logger.info(invocation.getArguments()[0].toString());
							return null;
						}
						else {
							return invocation.proceed();
						}
					}

				});
				return (Consumer<Object, Object>) pf.getProxy();
			}

		};
	}

	@KafkaListener(id = "foo", topics = "kgh537")
	public void listen(String in) {
		System.out.println(in);
	}

}

foo
2018-01-21 19:49:17.228 INFO 1584 --- [ foo-0-C-1] com.example.Kgh537Application : {kgh537-0=OffsetAndMetadata{offset=6, metadata=''}}

[dragontree101]

thanks very much !
but i think use aop is not very well.
if we can add some interface to commit function, i only implements this interface, i think it is more easy way to implement this feature

and i test your code, when set auto-commit is false, the log has print, if set auto-commit is true, not any log will been print.

whatever thanks your code! 👍