Java 类org.apache.kafka.clients.producer.internals.ProduceRequestResult 实例源码

/**
 * Adds the record to the list of sent records.
 *
 * @see #history()
 */
@Override
public synchronized Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
    verifyProducerState();
    int partition = 0;
    if (!this.cluster.partitionsForTopic(record.topic()).isEmpty())
        partition = partition(record, this.cluster);
    TopicPartition topicPartition = new TopicPartition(record.topic(), partition);
    ProduceRequestResult result = new ProduceRequestResult(topicPartition);
    FutureRecordMetadata future = new FutureRecordMetadata(result, 0, RecordBatch.NO_TIMESTAMP, 0L, 0, 0);
    long offset = nextOffset(topicPartition);
    Completion completion = new Completion(offset, new RecordMetadata(topicPartition, 0, offset,
            RecordBatch.NO_TIMESTAMP, Long.valueOf(0L), 0, 0), result, callback);

    if (!this.transactionInFlight)
        this.sent.add(record);
    else
        this.uncommittedSends.add(record);

    if (autoComplete)
        completion.complete(null);
    else
        this.completions.addLast(completion);

    return future;
}

/**
 * Test that waiting on a request that never completes times out
 */
@Test
public void testTimeout() throws Exception {
    ProduceRequestResult request = new ProduceRequestResult(topicPartition);
    FutureRecordMetadata future = new FutureRecordMetadata(request, relOffset,
            RecordBatch.NO_TIMESTAMP, 0L, 0, 0);
    assertFalse("Request is not completed", future.isDone());
    try {
        future.get(5, TimeUnit.MILLISECONDS);
        fail("Should have thrown exception.");
    } catch (TimeoutException e) { /* this is good */
    }

    request.set(baseOffset, RecordBatch.NO_TIMESTAMP, null);
    request.done();
    assertTrue(future.isDone());
    assertEquals(baseOffset + relOffset, future.get().offset());
}

/**
 * Adds the record to the list of sent records.
 *
 * @see #history()
 */
@Override
public synchronized Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback) {
    int partition = 0;
    if (this.cluster.partitionsForTopic(record.topic()) != null)
        partition = partition(record, this.cluster);
    ProduceRequestResult result = new ProduceRequestResult();
    FutureRecordMetadata future = new FutureRecordMetadata(result, 0, Record.NO_TIMESTAMP, 0, 0, 0);
    TopicPartition topicPartition = new TopicPartition(record.topic(), partition);
    long offset = nextOffset(topicPartition);
    Completion completion = new Completion(topicPartition, offset,
                                           new RecordMetadata(topicPartition, 0, offset, Record.NO_TIMESTAMP, 0, 0, 0),
                                           result, callback);
    this.sent.add(record);
    if (autoComplete)
        completion.complete(null);
    else
        this.completions.addLast(completion);
    return future;
}

/**
 * Test that waiting on a request that never completes times out
 */
@Test
public void testTimeout() throws Exception {
    ProduceRequestResult request = new ProduceRequestResult();
    FutureRecordMetadata future = new FutureRecordMetadata(request, relOffset,
                                                           Record.NO_TIMESTAMP, 0, 0, 0);
    assertFalse("Request is not completed", future.isDone());
    try {
        future.get(5, TimeUnit.MILLISECONDS);
        fail("Should have thrown exception.");
    } catch (TimeoutException e) { /* this is good */
    }

    request.done(topicPartition, baseOffset, null);
    assertTrue(future.isDone());
    assertEquals(baseOffset + relOffset, future.get().offset());
}

public Completion(long offset,
                  RecordMetadata metadata,
                  ProduceRequestResult result,
                  Callback callback) {
    this.metadata = metadata;
    this.offset = offset;
    this.result = result;
    this.callback = callback;
}

public ProduceRequestResult asyncRequest(final long baseOffset, final RuntimeException error, final long timeout) {
    final ProduceRequestResult request = new ProduceRequestResult(topicPartition);
    Thread thread = new Thread() {
        public void run() {
            try {
                sleep(timeout);
                request.set(baseOffset, RecordBatch.NO_TIMESTAMP, error);
                request.done();
            } catch (InterruptedException e) { }
        }
    };
    thread.start();
    return request;
}

public Completion(TopicPartition topicPartition,
                  long offset,
                  RecordMetadata metadata,
                  ProduceRequestResult result,
                  Callback callback) {
    this.metadata = metadata;
    this.offset = offset;
    this.result = result;
    this.callback = callback;
    this.topicPartition = topicPartition;
}

public ProduceRequestResult asyncRequest(final long baseOffset, final RuntimeException error, final long timeout) {
    final ProduceRequestResult request = new ProduceRequestResult();
    Thread thread = new Thread() {
        public void run() {
            try {
                sleep(timeout);
                request.done(topicPartition, baseOffset, error);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    };
    thread.start();
    return request;
}