@PrepareOnlyThisForTest(Metadata.class) @Test public void testInterceptorPartitionSetOnTooLargeRecord() throws Exception { Properties props = new Properties(); props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999"); props.setProperty(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, "1"); String topic = "topic"; ProducerRecord<String, String> record = new ProducerRecord<>(topic, "value"); KafkaProducer<String, String> producer = new KafkaProducer<>(props, new StringSerializer(), new StringSerializer()); Metadata metadata = PowerMock.createNiceMock(Metadata.class); MemberModifier.field(KafkaProducer.class, "metadata").set(producer, metadata); final Cluster cluster = new Cluster( "dummy", Collections.singletonList(new Node(0, "host1", 1000)), Arrays.asList(new PartitionInfo(topic, 0, null, null, null)), Collections.<String>emptySet(), Collections.<String>emptySet()); EasyMock.expect(metadata.fetch()).andReturn(cluster).once(); ProducerInterceptors interceptors = PowerMock.createMock(ProducerInterceptors.class); EasyMock.expect(interceptors.onSend(record)).andReturn(record); interceptors.onSendError(EasyMock.eq(record), EasyMock.<TopicPartition>notNull(), EasyMock.<Exception>notNull()); EasyMock.expectLastCall(); MemberModifier.field(KafkaProducer.class, "interceptors").set(producer, interceptors); PowerMock.replay(metadata); EasyMock.replay(interceptors); producer.send(record); EasyMock.verify(interceptors); }
@Override public Future<RecordMetadata> send(ProducerRecord<K, V> record) { return send(record, null); }
KafkaProducer implements Producer<K, V> { @Override public Future<RecordMetadata> send(ProducerRecord<K, V> record) { return send(record, null); } }
KafkaProducer implements Producer<K, V> { @Override public Future<RecordMetadata> send(ProducerRecord<K, V> record) { return send(record, null); } KafkaProducer(Map<String, Object> configs);  KafkaProducer(Map<String, Object> configs, Serializer<K> keySerializer, Serializer<V> valueSerializer);  KafkaProducer(Properties properties);  KafkaProducer(Properties properties, Serializer<K> keySerializer, Serializer<V> valueSerializer); @SuppressWarnings({"unchecked", "deprecation"}) private  KafkaProducer(ProducerConfig config, Serializer<K> keySerializer, Serializer<V> valueSerializer); }
KafkaProducer implements Producer<K, V> { @Override public Future<RecordMetadata> send(ProducerRecord<K, V> record) { return send(record, null); } KafkaProducer(Map<String, Object> configs);  KafkaProducer(Map<String, Object> configs, Serializer<K> keySerializer, Serializer<V> valueSerializer);  KafkaProducer(Properties properties);  KafkaProducer(Properties properties, Serializer<K> keySerializer, Serializer<V> valueSerializer); @SuppressWarnings({"unchecked", "deprecation"}) private  KafkaProducer(ProducerConfig config, Serializer<K> keySerializer, Serializer<V> valueSerializer); void initTransactions(); void beginTransaction(); void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
                                         String consumerGroupId); void commitTransaction(); void abortTransaction(); @Override Future<RecordMetadata> send(ProducerRecord<K, V> record); @Override // 发送消息，将消息放入RecordAccumulator暂存 Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback); @Override void flush(); @Override // 从Metadata中获取指定的topc分区信息 List<PartitionInfo> partitionsFor(String topic); @Override Map<MetricName, ? extends Metric> metrics(); @Override void close(); @Override void close(long timeout, TimeUnit timeUnit); }
KafkaProducer implements Producer<K, V> { @Override public Future<RecordMetadata> send(ProducerRecord<K, V> record) { return send(record, null); } KafkaProducer(Map<String, Object> configs);  KafkaProducer(Map<String, Object> configs, Serializer<K> keySerializer, Serializer<V> valueSerializer);  KafkaProducer(Properties properties);  KafkaProducer(Properties properties, Serializer<K> keySerializer, Serializer<V> valueSerializer); @SuppressWarnings({"unchecked", "deprecation"}) private  KafkaProducer(ProducerConfig config, Serializer<K> keySerializer, Serializer<V> valueSerializer); void initTransactions(); void beginTransaction(); void sendOffsetsToTransaction(Map<TopicPartition, OffsetAndMetadata> offsets,
                                         String consumerGroupId); void commitTransaction(); void abortTransaction(); @Override Future<RecordMetadata> send(ProducerRecord<K, V> record); @Override // 发送消息，将消息放入RecordAccumulator暂存 Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback); @Override void flush(); @Override // 从Metadata中获取指定的topc分区信息 List<PartitionInfo> partitionsFor(String topic); @Override Map<MetricName, ? extends Metric> metrics(); @Override void close(); @Override void close(long timeout, TimeUnit timeUnit);  }