test_class.declarator: mockProducer
test_class.modifier: private
test_class.original_string: private MockProducer<byte[], byte[]> mockProducer;
test_class.type: MockProducer<byte[], byte[]>
test_class.var_name: mockProducer
test_class.declarator: fiberProducer
test_class.modifier: private
test_class.original_string: private FiberKafkaProducer<byte[], byte[]> fiberProducer;
test_class.type: FiberKafkaProducer<byte[], byte[]>
test_class.var_name: fiberProducer
test_class.declarator: phaser
test_class.modifier: private
test_class.original_string: private co.paralleluniverse.strands.concurrent.Phaser phaser;
test_class.type: co.paralleluniverse.strands.concurrent.Phaser
test_class.var_name: phaser
test_class.file: comsat-kafka/src/test/java/co/paralleluniverse/fibers/kafka/FiberKafkaProducerTest.java
test_class.identifier: FiberKafkaProducerTest
test_case.body: @Test
      public void testSuccessfulSendWithCallback() throws ExecutionException, InterruptedException, TimeoutException {
          Fiber<Void> fiber = new Fiber<>(new SuspendableRunnable() {
              @Override
              public void run() throws SuspendExecution, InterruptedException {
  
                  final AtomicReference<RecordMetadata> callbackMetadata = new AtomicReference<>(null);
                  final AtomicReference<RecordMetadata> callbackMetadata2 = new AtomicReference<>(null);
  
                  Future<RecordMetadata> f = fiberProducer.send(new ProducerRecord<>("Topic", "Key".getBytes(), "Value".getBytes()),
                          new Callback() {
                              @Override
                              public void onCompletion(RecordMetadata metadata, Exception exception) {
                                  callbackMetadata.set(metadata);
                              }
                          });
                  Future<RecordMetadata> f2 = fiberProducer.send(new ProducerRecord<>("Topic", "Key".getBytes(), "Value".getBytes()),
                          new Callback() {
                              @Override
                              public void onCompletion(RecordMetadata metadata, Exception exception) {
                                  callbackMetadata2.set(metadata);
                              }
                          });
                  phaser.arriveAndAwaitAdvance();
  
                  try {
                      RecordMetadata recordMetadata = f.get();
  
                      assertEquals("Topic", recordMetadata.topic());
                      assertEquals(0, recordMetadata.offset());
                      assertEquals(0, recordMetadata.partition());
  
                      RecordMetadata recordMetadata2 = f2.get();
  
                      assertEquals("Topic", recordMetadata2.topic());
                      assertEquals(1, recordMetadata2.offset());
                      assertEquals(0, recordMetadata2.partition());
                  } catch (ExecutionException e) {
                      fail();
                  }
  
                  phaser.arriveAndAwaitAdvance();
  
                  assertEquals("Topic", callbackMetadata.get().topic());
                  assertEquals(0, callbackMetadata.get().offset());
                  assertEquals(0, callbackMetadata.get().partition());
  
                  assertEquals("Topic", callbackMetadata2.get().topic());
                  assertEquals(1, callbackMetadata2.get().offset());
                  assertEquals(0, callbackMetadata2.get().partition());
              }
          });
          fiber.start();
  
          phaser.arriveAndAwaitAdvance();
          mockProducer.completeNext();
          mockProducer.completeNext();
          phaser.arrive();
  
          fiber.join(5000, TimeUnit.MILLISECONDS);
      }
test_case.class_method_signature: FiberKafkaProducerTest.testSuccessfulSendWithCallback()
test_case.full_signature: @Test public void testSuccessfulSendWithCallback()
test_case.identifier: testSuccessfulSendWithCallback
test_case.invocations: send getBytes getBytes set send getBytes getBytes set arriveAndAwaitAdvance get assertEquals topic assertEquals offset assertEquals partition get assertEquals topic assertEquals offset assertEquals partition fail arriveAndAwaitAdvance assertEquals topic get assertEquals offset get assertEquals partition get assertEquals topic get assertEquals offset get assertEquals partition get start arriveAndAwaitAdvance completeNext completeNext arrive join
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testSuccessfulSendWithCallback()
focal_class.declarator: producer
focal_class.modifier: private final
focal_class.original_string: private final Producer<K, V> producer;
focal_class.type: Producer<K, V>
focal_class.var_name: producer
focal_class.file: comsat-kafka/src/main/java/co/paralleluniverse/fibers/kafka/FiberKafkaProducer.java
focal_class.identifier: FiberKafkaProducer
focal_class.interfaces: implements Producer<K, V>
focal_class.class_method_signature: FiberKafkaProducer.FiberKafkaProducer(Producer<K, V> producer)
focal_class.full_signature: public  FiberKafkaProducer(Producer<K, V> producer)
focal_class.identifier: FiberKafkaProducer
focal_class.modifiers: public
focal_class.parameters: (Producer<K, V> producer)
focal_class.signature:  FiberKafkaProducer(Producer<K, V> producer)
focal_class.class_method_signature: FiberKafkaProducer.send(ProducerRecord<K, V> record)
focal_class.full_signature: @Override public Future<RecordMetadata> send(ProducerRecord<K, V> record)
focal_class.identifier: send
focal_class.modifiers: @Override public
focal_class.parameters: (ProducerRecord<K, V> record)
focal_class.return: Future<RecordMetadata>
focal_class.signature: Future<RecordMetadata> send(ProducerRecord<K, V> record)
focal_class.class_method_signature: FiberKafkaProducer.send(ProducerRecord<K, V> record, Callback callback)
focal_class.full_signature: @Override public Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback)
focal_class.identifier: send
focal_class.modifiers: @Override public
focal_class.parameters: (ProducerRecord<K, V> record, Callback callback)
focal_class.return: Future<RecordMetadata>
focal_class.signature: Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback)
focal_class.class_method_signature: FiberKafkaProducer.flush()
focal_class.full_signature: @Override public void flush()
focal_class.identifier: flush
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void flush()
focal_class.class_method_signature: FiberKafkaProducer.partitionsFor(String topic)
focal_class.full_signature: @Override public List<PartitionInfo> partitionsFor(String topic)
focal_class.identifier: partitionsFor
focal_class.modifiers: @Override public
focal_class.parameters: (String topic)
focal_class.return: List<PartitionInfo>
focal_class.signature: List<PartitionInfo> partitionsFor(String topic)
focal_class.class_method_signature: FiberKafkaProducer.metrics()
focal_class.full_signature: @Override public Map<MetricName, ? extends Metric> metrics()
focal_class.identifier: metrics
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: Map<MetricName, ? extends Metric>
focal_class.signature: Map<MetricName, ? extends Metric> metrics()
focal_class.class_method_signature: FiberKafkaProducer.close()
focal_class.full_signature: @Override public void close()
focal_class.identifier: close
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void close()
focal_class.class_method_signature: FiberKafkaProducer.close(long timeout, TimeUnit unit)
focal_class.full_signature: @Override public void close(long timeout, TimeUnit unit)
focal_class.identifier: close
focal_class.modifiers: @Override public
focal_class.parameters: (long timeout, TimeUnit unit)
focal_class.return: void
focal_class.signature: void close(long timeout, TimeUnit unit)
focal_method.body: @Override
      public Future<RecordMetadata> send(ProducerRecord<K, V> record) {
          return send(record, null);
      }
focal_method.class_method_signature: FiberKafkaProducer.send(ProducerRecord<K, V> record)
focal_method.full_signature: @Override public Future<RecordMetadata> send(ProducerRecord<K, V> record)
focal_method.identifier: send
focal_method.invocations: send
focal_method.modifiers: @Override public
focal_method.parameters: (ProducerRecord<K, V> record)
focal_method.return: Future<RecordMetadata>
focal_method.signature: Future<RecordMetadata> send(ProducerRecord<K, V> record)