### Question:
KafkaProducerWrapper implements Closeable, Flushable { public void sendSynchronously(ProducerRecord<K, T> record) throws IOException { sendSynchronously(Collections.singletonList(record)); } KafkaProducerWrapper(Producer<K, T> kafkaProducer); void send(ProducerRecord<K, T> record); void sendSynchronously(ProducerRecord<K, T> record); void sendSynchronously(List<ProducerRecord<K, T>> records); @Override void flush(); @Override void close();  }### Answer:
@Test public void testSynchronous_messageTooLarge() throws IOException { kafkaAdminClient.createTopic(topic, 4, 1, new Properties()); Properties props = KafkaTests.getProps(); props.setProperty(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.setProperty(VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName()); props.setProperty(ProducerConfig.BATCH_SIZE_CONFIG, "10000"); props.setProperty(ProducerConfig.LINGER_MS_CONFIG, "60000"); StringBuilder bldr = new StringBuilder(); for(int i = 0; i < 100000; i++) { bldr.append(i); } List<ProducerRecord<String, String>> records = new ArrayList<>(2); records.add(new ProducerRecord<>(topic, "key", bldr.toString())); records.add(new ProducerRecord<>(topic, "key2", "small")); boolean caughtRecordTooLargeException = false; try (KafkaProducerWrapper<String, String> producer = new KafkaProducerWrapper<>(new KafkaProducer<>(props))) { producer.sendSynchronously(records); } catch (KafkaExecutionException e) { Throwable cause = e.getCause(); assertThat(cause, instanceOf(ExecutionException.class)); Throwable cause2 = cause.getCause(); assertThat(cause2, instanceOf(RecordTooLargeException.class)); caughtRecordTooLargeException = true; } assertThat(caughtRecordTooLargeException, is(true)); }