@Test public void testBackoffMsUpdatesReadTaskExpiry() throws Exception { Properties props = setUpProperties(); props.put(KafkaRestConfig.CONSUMER_ITERATOR_BACKOFF_MS_CONFIG, "1000"); config = new KafkaRestConfig(props, new SystemTime()); consumerManager = new KafkaConsumerManager(config, consumerFactory); consumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST, groupName); bootstrapConsumer(consumer); consumerManager.readRecords(groupName, consumer.cid(), BinaryKafkaConsumerState.class, -1, Long.MAX_VALUE, new ConsumerReadCallback<ByteString, ByteString>() { @Override public void onCompletion( List<ConsumerRecord<ByteString, ByteString>> records, Exception e) { actualException = e; actualRecords = records; sawCallback = true; } }); Thread.sleep(100); KafkaConsumerManager.RunnableReadTask readTask = consumerManager.delayedReadTasks.peek(); if (readTask == null) { fail("Could not get read task in time. It should not be null"); } long delay = readTask.getDelay(TimeUnit.MILLISECONDS); assertTrue(delay < 1000); assertTrue(delay > 700); }
public <KafkaKeyT, KafkaValueT, ClientKeyT, ClientValueT> void readRecords( final String group, final String instance, Class<? extends KafkaConsumerState<KafkaKeyT, KafkaValueT, ClientKeyT, ClientValueT>> consumerStateType, final long timeout, final long maxBytes, final ConsumerReadCallback<ClientKeyT, ClientValueT> callback ) { final KafkaConsumerState state; try { state = getConsumerInstance(group, instance); } catch (RestNotFoundException e) { callback.onCompletion(null, e); return; } if (!consumerStateType.isInstance(state)) { callback.onCompletion(null, Errors.consumerFormatMismatch()); return; } final KafkaConsumerReadTask task = new KafkaConsumerReadTask<KafkaKeyT, KafkaValueT, ClientKeyT, ClientValueT>( state, timeout, maxBytes, callback ); executor.submit(new RunnableReadTask(new ReadTaskState(task, state, callback))); }
KafkaConsumerManager { public <KafkaKeyT, KafkaValueT, ClientKeyT, ClientValueT> void readRecords( final String group, final String instance, Class<? extends KafkaConsumerState<KafkaKeyT, KafkaValueT, ClientKeyT, ClientValueT>> consumerStateType, final long timeout, final long maxBytes, final ConsumerReadCallback<ClientKeyT, ClientValueT> callback ) { final KafkaConsumerState state; try { state = getConsumerInstance(group, instance); } catch (RestNotFoundException e) { callback.onCompletion(null, e); return; } if (!consumerStateType.isInstance(state)) { callback.onCompletion(null, Errors.consumerFormatMismatch()); return; } final KafkaConsumerReadTask task = new KafkaConsumerReadTask<KafkaKeyT, KafkaValueT, ClientKeyT, ClientValueT>( state, timeout, maxBytes, callback ); executor.submit(new RunnableReadTask(new ReadTaskState(task, state, callback))); } }
KafkaConsumerManager { public <KafkaKeyT, KafkaValueT, ClientKeyT, ClientValueT> void readRecords( final String group, final String instance, Class<? extends KafkaConsumerState<KafkaKeyT, KafkaValueT, ClientKeyT, ClientValueT>> consumerStateType, final long timeout, final long maxBytes, final ConsumerReadCallback<ClientKeyT, ClientValueT> callback ) { final KafkaConsumerState state; try { state = getConsumerInstance(group, instance); } catch (RestNotFoundException e) { callback.onCompletion(null, e); return; } if (!consumerStateType.isInstance(state)) { callback.onCompletion(null, Errors.consumerFormatMismatch()); return; } final KafkaConsumerReadTask task = new KafkaConsumerReadTask<KafkaKeyT, KafkaValueT, ClientKeyT, ClientValueT>( state, timeout, maxBytes, callback ); executor.submit(new RunnableReadTask(new ReadTaskState(task, state, callback))); } KafkaConsumerManager(final KafkaRestConfig config);   KafkaConsumerManager(KafkaRestConfig config, KafkaConsumerFactory consumerFactory); }
KafkaConsumerManager { public <KafkaKeyT, KafkaValueT, ClientKeyT, ClientValueT> void readRecords( final String group, final String instance, Class<? extends KafkaConsumerState<KafkaKeyT, KafkaValueT, ClientKeyT, ClientValueT>> consumerStateType, final long timeout, final long maxBytes, final ConsumerReadCallback<ClientKeyT, ClientValueT> callback ) { final KafkaConsumerState state; try { state = getConsumerInstance(group, instance); } catch (RestNotFoundException e) { callback.onCompletion(null, e); return; } if (!consumerStateType.isInstance(state)) { callback.onCompletion(null, Errors.consumerFormatMismatch()); return; } final KafkaConsumerReadTask task = new KafkaConsumerReadTask<KafkaKeyT, KafkaValueT, ClientKeyT, ClientValueT>( state, timeout, maxBytes, callback ); executor.submit(new RunnableReadTask(new ReadTaskState(task, state, callback))); } KafkaConsumerManager(final KafkaRestConfig config);   KafkaConsumerManager(KafkaRestConfig config, KafkaConsumerFactory consumerFactory); String createConsumer(String group, ConsumerInstanceConfig instanceConfig); void readRecords(
      final String group,
      final String instance,
      Class<? extends KafkaConsumerState<KafkaKeyT, KafkaValueT, ClientKeyT, ClientValueT>>
          consumerStateType,
      final long timeout,
      final long maxBytes,
      final ConsumerReadCallback<ClientKeyT, ClientValueT> callback
  ); Future commitOffsets(
      String group, String instance, final String async,
      final ConsumerOffsetCommitRequest offsetCommitRequest, final CommitCallback callback
  ); ConsumerCommittedResponse committed(
      String group,
      String instance,
      ConsumerCommittedRequest request
  ); long getBeginningOffset(String topic, int partition); long getEndOffset(String topic, int partition); Optional<Long> getOffsetForTime(
      String topic, int partition, Instant timestamp); void deleteConsumer(String group, String instance); void subscribe(String group, String instance, ConsumerSubscriptionRecord subscription); void unsubscribe(String group, String instance); ConsumerSubscriptionResponse subscription(String group, String instance); void seekToBeginning(String group, String instance, ConsumerSeekToRequest seekToRequest); void seekToEnd(String group, String instance, ConsumerSeekToRequest seekToRequest); void seekToOffset(
      String group,
      String instance,
      ConsumerSeekToOffsetRequest seekToOffsetRequest
  ); void assign(String group, String instance, ConsumerAssignmentRequest assignmentRequest); ConsumerAssignmentResponse assignment(String group, String instance); void shutdown(); }
KafkaConsumerManager { public <KafkaKeyT, KafkaValueT, ClientKeyT, ClientValueT> void readRecords( final String group, final String instance, Class<? extends KafkaConsumerState<KafkaKeyT, KafkaValueT, ClientKeyT, ClientValueT>> consumerStateType, final long timeout, final long maxBytes, final ConsumerReadCallback<ClientKeyT, ClientValueT> callback ) { final KafkaConsumerState state; try { state = getConsumerInstance(group, instance); } catch (RestNotFoundException e) { callback.onCompletion(null, e); return; } if (!consumerStateType.isInstance(state)) { callback.onCompletion(null, Errors.consumerFormatMismatch()); return; } final KafkaConsumerReadTask task = new KafkaConsumerReadTask<KafkaKeyT, KafkaValueT, ClientKeyT, ClientValueT>( state, timeout, maxBytes, callback ); executor.submit(new RunnableReadTask(new ReadTaskState(task, state, callback))); } KafkaConsumerManager(final KafkaRestConfig config);   KafkaConsumerManager(KafkaRestConfig config, KafkaConsumerFactory consumerFactory); String createConsumer(String group, ConsumerInstanceConfig instanceConfig); void readRecords(
      final String group,
      final String instance,
      Class<? extends KafkaConsumerState<KafkaKeyT, KafkaValueT, ClientKeyT, ClientValueT>>
          consumerStateType,
      final long timeout,
      final long maxBytes,
      final ConsumerReadCallback<ClientKeyT, ClientValueT> callback
  ); Future commitOffsets(
      String group, String instance, final String async,
      final ConsumerOffsetCommitRequest offsetCommitRequest, final CommitCallback callback
  ); ConsumerCommittedResponse committed(
      String group,
      String instance,
      ConsumerCommittedRequest request
  ); long getBeginningOffset(String topic, int partition); long getEndOffset(String topic, int partition); Optional<Long> getOffsetForTime(
      String topic, int partition, Instant timestamp); void deleteConsumer(String group, String instance); void subscribe(String group, String instance, ConsumerSubscriptionRecord subscription); void unsubscribe(String group, String instance); ConsumerSubscriptionResponse subscription(String group, String instance); void seekToBeginning(String group, String instance, ConsumerSeekToRequest seekToRequest); void seekToEnd(String group, String instance, ConsumerSeekToRequest seekToRequest); void seekToOffset(
      String group,
      String instance,
      ConsumerSeekToOffsetRequest seekToOffsetRequest
  ); void assign(String group, String instance, ConsumerAssignmentRequest assignmentRequest); ConsumerAssignmentResponse assignment(String group, String instance); void shutdown();  }