TaskContextImpl implements TaskContext { @Override public void setStartingOffset(SystemStreamPartition systemStreamPartition, String offset) { this.offsetManager.setStartingOffset(this.taskModel.getTaskName(), systemStreamPartition, offset); } TaskContextImpl(TaskModel taskModel,
      MetricsRegistry taskMetricsRegistry,
      Function<String, KeyValueStore> keyValueStoreProvider,
      TableManager tableManager,
      CallbackScheduler callbackScheduler,
      OffsetManager offsetManager,
      JobModel jobModel,
      StreamMetadataCache streamMetadataCache); @Override TaskModel getTaskModel(); @Override MetricsRegistry getTaskMetricsRegistry(); @Override KeyValueStore getStore(String storeName); @Override ReadWriteTable<K, V> getTable(String tableId); @Override CallbackScheduler getCallbackScheduler(); @Override void setStartingOffset(SystemStreamPartition systemStreamPartition, String offset); JobModel getJobModel(); StreamMetadataCache getStreamMetadataCache();  }
@Test public void testSetStartingOffset() { SystemStreamPartition ssp = new SystemStreamPartition("mySystem", "myStream", new Partition(0)); taskContext.setStartingOffset(ssp, "123"); verify(offsetManager).setStartingOffset(TASK_NAME, ssp, "123"); }