class KafkaItemWriter<K, T> extends KeyValueItemWriter<K, T> {

	protected KafkaTemplate<K, T> kafkaTemplate;

	protected final List<CompletableFuture<SendResult<K, T>>> completableFutures = new ArrayList<>();

	private long timeout = -1;

	@Override
	protected void writeKeyValue(K key, T value) {<FILL_FUNCTION_BODY>}

	@Override
	protected void flush() throws Exception {
		this.kafkaTemplate.flush();
		for (var future : this.completableFutures) {
			if (this.timeout >= 0) {
				future.get(this.timeout, TimeUnit.MILLISECONDS);
			}
			else {
				future.get();
			}
		}
		this.completableFutures.clear();
	}

	@Override
	protected void init() {
		Assert.state(this.kafkaTemplate != null, "KafkaTemplate must not be null.");
		Assert.state(this.kafkaTemplate.getDefaultTopic() != null, "KafkaTemplate must have the default topic set.");
	}

	/**
	 * Set the {@link KafkaTemplate} to use.
	 * @param kafkaTemplate to use
	 */
	public void setKafkaTemplate(KafkaTemplate<K, T> kafkaTemplate) {
		this.kafkaTemplate = kafkaTemplate;
	}

	/**
	 * The time limit to wait when flushing items to Kafka.
	 * @param timeout milliseconds to wait, defaults to -1 (no timeout).
	 * @since 4.3.2
	 */
	public void setTimeout(long timeout) {
		this.timeout = timeout;
	}

}

		if (this.delete) {
			this.completableFutures.add(this.kafkaTemplate.sendDefault(key, null));
		}
		else {
			this.completableFutures.add(this.kafkaTemplate.sendDefault(key, value));
		}
	
<methods>public non-sealed void <init>() ,public void afterPropertiesSet() throws java.lang.Exception,public void setDelete(boolean) ,public void setItemKeyMapper(Converter<T,K>) ,public void write(Chunk<? extends T>) throws java.lang.Exception<variables>protected boolean delete,protected Converter<T,K> itemKeyMapper