LocalTableProvider extends BaseTableProvider { @Override public void init(Context context) { super.init(context); Preconditions.checkNotNull(this.context, "Must specify context for local tables."); kvStore = this.context.getTaskContext().getStore(tableId); Preconditions.checkNotNull(kvStore, String.format( "Backing store for table %s was not injected by SamzaContainer", tableId)); logger.info("Initialized backing store for table " + tableId); } LocalTableProvider(String tableId); @Override void init(Context context); @Override ReadWriteTable getTable();  }
@Test public void testInit() { Context context = mock(Context.class); JobContext jobContext = mock(JobContext.class); when(context.getJobContext()).thenReturn(jobContext); when(jobContext.getConfig()).thenReturn(new MapConfig()); ContainerContext containerContext = mock(ContainerContext.class); when(context.getContainerContext()).thenReturn(containerContext); when(containerContext.getContainerMetricsRegistry()).thenReturn(new NoOpMetricsRegistry()); TaskContext taskContext = mock(TaskContext.class); when(context.getTaskContext()).thenReturn(taskContext); when(taskContext.getStore(any())).thenReturn(mock(KeyValueStore.class)); TableProvider tableProvider = createTableProvider("t1"); tableProvider.init(context); Assert.assertNotNull(tableProvider.getTable()); }