test_class.declarator: server
test_class.modifier: private static
test_class.original_string: private static NetworkServerControl server;
test_class.type: NetworkServerControl
test_class.var_name: server
test_class.declarator: tempDir
test_class.modifier: private static
test_class.original_string: private static File tempDir;
test_class.type: File
test_class.var_name: tempDir
test_class.declarator: conf
test_class.modifier: private static
test_class.original_string: private static DbBackendConfig conf;
test_class.type: DbBackendConfig
test_class.var_name: conf
test_class.declarator: url1
test_class.modifier: private static
test_class.original_string: private static String url1;
test_class.type: String
test_class.var_name: url1
test_class.declarator: url2
test_class.modifier: private static
test_class.original_string: private static String url2;
test_class.type: String
test_class.var_name: url2
test_class.file: flink-contrib/flink-streaming-contrib/src/test/java/org/apache/flink/contrib/streaming/state/DbStateBackendTest.java
test_class.identifier: DbStateBackendTest
test_case.body: @Test
  	@SuppressWarnings("unchecked,rawtypes")
  	public void testListState() {
  		File tempDir = new File(ConfigConstants.DEFAULT_TASK_MANAGER_TMP_PATH, UUID.randomUUID().toString());
  		try {
  			FsStateBackend fileBackend = new FsStateBackend(localFileUri(tempDir));
  
  			DbStateBackend backend = new DbStateBackend(conf, fileBackend);
  
  			Environment env = new DummyEnvironment("test", 2, 0);
  
  			backend.initializeForJob(env, "dummy_test_kv_list", IntSerializer.INSTANCE);
  
  			ListStateDescriptor<String> kvId = new ListStateDescriptor<>("id", StringSerializer.INSTANCE);
  			ListState<String> state = backend.getPartitionedState(null, VoidSerializer.INSTANCE, kvId);
  
  			@SuppressWarnings("unchecked")
  			KvState<Integer, Void, ListState<String>, ListStateDescriptor<String>, DbStateBackend> kv =
  				(KvState<Integer, Void, ListState<String>, ListStateDescriptor<String>, DbStateBackend>) state;
  
  			Joiner joiner = Joiner.on(",");
  			// some modifications to the state
  			kv.setCurrentKey(1);
  			assertEquals("", joiner.join(state.get()));
  			state.add("1");
  			kv.setCurrentKey(2);
  			assertEquals("", joiner.join(state.get()));
  			state.add("2");
  			kv.setCurrentKey(1);
  			assertEquals("1", joiner.join(state.get()));
  
  			// draw a snapshot
  			KvStateSnapshot<Integer, Void, ListState<String>, ListStateDescriptor<String>, DbStateBackend> snapshot1 =
  				kv.snapshot(682375462378L, 2);
  
  			// make some more modifications
  			kv.setCurrentKey(1);
  			state.add("u1");
  			kv.setCurrentKey(2);
  			state.add("u2");
  			kv.setCurrentKey(3);
  			state.add("u3");
  
  			// draw another snapshot
  			KvStateSnapshot<Integer, Void, ListState<String>, ListStateDescriptor<String>, DbStateBackend> snapshot2 =
  				kv.snapshot(682375462379L, 4);
  
  			// validate the original state
  			kv.setCurrentKey(1);
  			assertEquals("1,u1", joiner.join(state.get()));
  			kv.setCurrentKey(2);
  			assertEquals("2,u2", joiner.join(state.get()));
  			kv.setCurrentKey(3);
  			assertEquals("u3", joiner.join(state.get()));
  
  			kv.dispose();
  
  			// restore the second snapshot and validate it
  			KvState<Integer, Void, ListState<String>, ListStateDescriptor<String>, DbStateBackend> restored2 = snapshot2.restoreState(
  				backend,
  				IntSerializer.INSTANCE,
  				this.getClass().getClassLoader(), 20);
  
  			@SuppressWarnings("unchecked")
  			ListState<String> restored2State = (ListState<String>) restored2;
  
  			restored2.setCurrentKey(1);
  			assertEquals("1,u1", joiner.join(restored2State.get()));
  			restored2.setCurrentKey(2);
  			assertEquals("2,u2", joiner.join(restored2State.get()));
  			restored2.setCurrentKey(3);
  			assertEquals("u3", joiner.join(restored2State.get()));
  		}
  		catch (Exception e) {
  			e.printStackTrace();
  			fail(e.getMessage());
  		}
  	}
test_case.class_method_signature: DbStateBackendTest.testListState()
test_case.full_signature: @Test @SuppressWarnings("unchecked,rawtypes") public void testListState()
test_case.identifier: testListState
test_case.invocations: toString randomUUID localFileUri initializeForJob getPartitionedState on setCurrentKey assertEquals join get add setCurrentKey assertEquals join get add setCurrentKey assertEquals join get snapshot setCurrentKey add setCurrentKey add setCurrentKey add snapshot setCurrentKey assertEquals join get setCurrentKey assertEquals join get setCurrentKey assertEquals join get dispose restoreState getClassLoader getClass setCurrentKey assertEquals join get setCurrentKey assertEquals join get setCurrentKey assertEquals join get printStackTrace fail getMessage
test_case.modifiers: @Test @SuppressWarnings("unchecked,rawtypes") public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testListState()
focal_class.declarator: serialVersionUID = 1L
focal_class.modifier: private static final
focal_class.original_string: private static final long serialVersionUID = 1L;
focal_class.type: long
focal_class.var_name: serialVersionUID
focal_class.declarator: LOG = LoggerFactory.getLogger(DbStateBackend.class)
focal_class.modifier: private static final
focal_class.original_string: private static final Logger LOG = LoggerFactory.getLogger(DbStateBackend.class);
focal_class.type: Logger
focal_class.var_name: LOG
focal_class.declarator: rnd
focal_class.modifier: private
focal_class.original_string: private Random rnd;
focal_class.type: Random
focal_class.var_name: rnd
focal_class.declarator: env
focal_class.modifier: private transient
focal_class.original_string: private transient Environment env;
focal_class.type: Environment
focal_class.var_name: env
focal_class.declarator: jobId
focal_class.modifier: private transient
focal_class.original_string: private transient String jobId;
focal_class.type: String
focal_class.var_name: jobId
focal_class.declarator: dbConfig
focal_class.modifier: private final
focal_class.original_string: private final DbBackendConfig dbConfig;
focal_class.type: DbBackendConfig
focal_class.var_name: dbConfig
focal_class.declarator: dbAdapter
focal_class.modifier: private final
focal_class.original_string: private final DbAdapter dbAdapter;
focal_class.type: DbAdapter
focal_class.var_name: dbAdapter
focal_class.declarator: connections
focal_class.modifier: private
focal_class.original_string: private ShardedConnection connections;
focal_class.type: ShardedConnection
focal_class.var_name: connections
focal_class.declarator: numSqlRetries
focal_class.modifier: private final
focal_class.original_string: private final int numSqlRetries;
focal_class.type: int
focal_class.var_name: numSqlRetries
focal_class.declarator: sqlRetrySleep
focal_class.modifier: private final
focal_class.original_string: private final int sqlRetrySleep;
focal_class.type: int
focal_class.var_name: sqlRetrySleep
focal_class.declarator: insertStatement
focal_class.modifier: private transient
focal_class.original_string: private transient PreparedStatement insertStatement;
focal_class.type: PreparedStatement
focal_class.var_name: insertStatement
focal_class.declarator: operatorIdentifier
focal_class.modifier: private
focal_class.original_string: private String operatorIdentifier;
focal_class.type: String
focal_class.var_name: operatorIdentifier
focal_class.declarator: nonPartitionedStateBackend = null
focal_class.modifier: private
focal_class.original_string: private AbstractStateBackend nonPartitionedStateBackend = null;
focal_class.type: AbstractStateBackend
focal_class.var_name: nonPartitionedStateBackend
focal_class.file: flink-contrib/flink-streaming-contrib/src/main/java/org/apache/flink/contrib/streaming/state/DbStateBackend.java
focal_class.identifier: DbStateBackend
focal_class.class_method_signature: DbStateBackend.DbStateBackend(DbBackendConfig backendConfig)
focal_class.full_signature: public  DbStateBackend(DbBackendConfig backendConfig)
focal_class.identifier: DbStateBackend
focal_class.modifiers: public
focal_class.parameters: (DbBackendConfig backendConfig)
focal_class.signature:  DbStateBackend(DbBackendConfig backendConfig)
focal_class.class_method_signature: DbStateBackend.DbStateBackend(DbBackendConfig backendConfig, AbstractStateBackend backend)
focal_class.full_signature: public  DbStateBackend(DbBackendConfig backendConfig, AbstractStateBackend backend)
focal_class.identifier: DbStateBackend
focal_class.modifiers: public
focal_class.parameters: (DbBackendConfig backendConfig, AbstractStateBackend backend)
focal_class.signature:  DbStateBackend(DbBackendConfig backendConfig, AbstractStateBackend backend)
focal_class.class_method_signature: DbStateBackend.getConnections()
focal_class.full_signature: public ShardedConnection getConnections()
focal_class.identifier: getConnections
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: ShardedConnection
focal_class.signature: ShardedConnection getConnections()
focal_class.class_method_signature: DbStateBackend.isInitialized()
focal_class.full_signature: public boolean isInitialized()
focal_class.identifier: isInitialized
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: boolean
focal_class.signature: boolean isInitialized()
focal_class.class_method_signature: DbStateBackend.getEnvironment()
focal_class.full_signature: public Environment getEnvironment()
focal_class.identifier: getEnvironment
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: Environment
focal_class.signature: Environment getEnvironment()
focal_class.class_method_signature: DbStateBackend.getConfiguration()
focal_class.full_signature: public DbBackendConfig getConfiguration()
focal_class.identifier: getConfiguration
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: DbBackendConfig
focal_class.signature: DbBackendConfig getConfiguration()
focal_class.class_method_signature: DbStateBackend.checkpointStateSerializable(final S state, final long checkpointID,
  			final long timestamp)
focal_class.full_signature: @Override public StateHandle<S> checkpointStateSerializable(final S state, final long checkpointID,
  			final long timestamp)
focal_class.identifier: checkpointStateSerializable
focal_class.modifiers: @Override public
focal_class.parameters: (final S state, final long checkpointID,
  			final long timestamp)
focal_class.return: StateHandle<S>
focal_class.signature: StateHandle<S> checkpointStateSerializable(final S state, final long checkpointID,
  			final long timestamp)
focal_class.class_method_signature: DbStateBackend.createCheckpointStateOutputStream(long checkpointID, long timestamp)
focal_class.full_signature: @Override public CheckpointStateOutputStream createCheckpointStateOutputStream(long checkpointID, long timestamp)
focal_class.identifier: createCheckpointStateOutputStream
focal_class.modifiers: @Override public
focal_class.parameters: (long checkpointID, long timestamp)
focal_class.return: CheckpointStateOutputStream
focal_class.signature: CheckpointStateOutputStream createCheckpointStateOutputStream(long checkpointID, long timestamp)
focal_class.class_method_signature: DbStateBackend.createValueState(TypeSerializer<N> namespaceSerializer,
  		ValueStateDescriptor<T> stateDesc)
focal_class.full_signature: @Override protected ValueState<T> createValueState(TypeSerializer<N> namespaceSerializer,
  		ValueStateDescriptor<T> stateDesc)
focal_class.identifier: createValueState
focal_class.modifiers: @Override protected
focal_class.parameters: (TypeSerializer<N> namespaceSerializer,
  		ValueStateDescriptor<T> stateDesc)
focal_class.return: ValueState<T>
focal_class.signature: ValueState<T> createValueState(TypeSerializer<N> namespaceSerializer,
  		ValueStateDescriptor<T> stateDesc)
focal_class.class_method_signature: DbStateBackend.createListState(TypeSerializer<N> namespaceSerializer,
  		ListStateDescriptor<T> stateDesc)
focal_class.full_signature: @Override protected ListState<T> createListState(TypeSerializer<N> namespaceSerializer,
  		ListStateDescriptor<T> stateDesc)
focal_class.identifier: createListState
focal_class.modifiers: @Override protected
focal_class.parameters: (TypeSerializer<N> namespaceSerializer,
  		ListStateDescriptor<T> stateDesc)
focal_class.return: ListState<T>
focal_class.signature: ListState<T> createListState(TypeSerializer<N> namespaceSerializer,
  		ListStateDescriptor<T> stateDesc)
focal_class.class_method_signature: DbStateBackend.createReducingState(TypeSerializer<N> namespaceSerializer,
  		ReducingStateDescriptor<T> stateDesc)
focal_class.full_signature: @Override @SuppressWarnings("unchecked") protected ReducingState<T> createReducingState(TypeSerializer<N> namespaceSerializer,
  		ReducingStateDescriptor<T> stateDesc)
focal_class.identifier: createReducingState
focal_class.modifiers: @Override @SuppressWarnings("unchecked") protected
focal_class.parameters: (TypeSerializer<N> namespaceSerializer,
  		ReducingStateDescriptor<T> stateDesc)
focal_class.return: ReducingState<T>
focal_class.signature: ReducingState<T> createReducingState(TypeSerializer<N> namespaceSerializer,
  		ReducingStateDescriptor<T> stateDesc)
focal_class.class_method_signature: DbStateBackend.createFoldingState(TypeSerializer<N> namespaceSerializer,
  			FoldingStateDescriptor<T, ACC> stateDesc)
focal_class.full_signature: @Override protected FoldingState<T, ACC> createFoldingState(TypeSerializer<N> namespaceSerializer,
  			FoldingStateDescriptor<T, ACC> stateDesc)
focal_class.identifier: createFoldingState
focal_class.modifiers: @Override protected
focal_class.parameters: (TypeSerializer<N> namespaceSerializer,
  			FoldingStateDescriptor<T, ACC> stateDesc)
focal_class.return: FoldingState<T, ACC>
focal_class.signature: FoldingState<T, ACC> createFoldingState(TypeSerializer<N> namespaceSerializer,
  			FoldingStateDescriptor<T, ACC> stateDesc)
focal_class.class_method_signature: DbStateBackend.initializeForJob(final Environment env,
  		String operatorIdentifier,
  		TypeSerializer<?> keySerializer)
focal_class.full_signature: @Override public void initializeForJob(final Environment env,
  		String operatorIdentifier,
  		TypeSerializer<?> keySerializer)
focal_class.identifier: initializeForJob
focal_class.modifiers: @Override public
focal_class.parameters: (final Environment env,
  		String operatorIdentifier,
  		TypeSerializer<?> keySerializer)
focal_class.return: void
focal_class.signature: void initializeForJob(final Environment env,
  		String operatorIdentifier,
  		TypeSerializer<?> keySerializer)
focal_class.class_method_signature: DbStateBackend.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: DbStateBackend.disposeAllStateForCurrentJob()
focal_class.full_signature: @Override public void disposeAllStateForCurrentJob()
focal_class.identifier: disposeAllStateForCurrentJob
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void disposeAllStateForCurrentJob()
focal_class.superclass: extends AbstractStateBackend
focal_method.body: @Override
  	public void initializeForJob(final Environment env,
  		String operatorIdentifier,
  		TypeSerializer<?> keySerializer) throws Exception {
  		super.initializeForJob(env, operatorIdentifier, keySerializer);
  
  		this.operatorIdentifier = operatorIdentifier;
  
  		this.rnd = new Random();
  		this.env = env;
  		this.jobId = env.getJobID().toString().substring(0, 16);
  
  		connections = dbConfig.createShardedConnection();
  
  		// We want the most light-weight transaction isolation level as we don't
  		// have conflicting reads/writes. We just want to be able to roll back
  		// batch inserts for k-v snapshots. This requirement might be removed in
  		// the future.
  		connections.setTransactionIsolation(Connection.TRANSACTION_READ_UNCOMMITTED);
  
  		// If we have a different backend for non-partitioned states we
  		// initialize that, otherwise create tables for storing the checkpoints.
  		//
  		// Currently all non-partitioned states are written to the first
  		// database shard
  		if (nonPartitionedStateBackend == null) {
  			insertStatement = retry(new Callable<PreparedStatement>() {
  				public PreparedStatement call() throws SQLException {
  					dbAdapter.createCheckpointsTable(jobId, getConnections().getFirst());
  					return dbAdapter.prepareCheckpointInsert(jobId,
  							getConnections().getFirst());
  				}
  			}, numSqlRetries, sqlRetrySleep);
  		} else {
  			nonPartitionedStateBackend.initializeForJob(env, operatorIdentifier, keySerializer);
  		}
  
  		if (LOG.isDebugEnabled()) {
  			LOG.debug("Database state backend successfully initialized");
  		}
  	}
focal_method.class_method_signature: DbStateBackend.initializeForJob(final Environment env,
  		String operatorIdentifier,
  		TypeSerializer<?> keySerializer)
focal_method.full_signature: @Override public void initializeForJob(final Environment env,
  		String operatorIdentifier,
  		TypeSerializer<?> keySerializer)
focal_method.identifier: initializeForJob
focal_method.invocations: initializeForJob substring toString getJobID createShardedConnection setTransactionIsolation retry createCheckpointsTable getFirst getConnections prepareCheckpointInsert getFirst getConnections initializeForJob isDebugEnabled debug
focal_method.modifiers: @Override public
focal_method.parameters: (final Environment env,
  		String operatorIdentifier,
  		TypeSerializer<?> keySerializer)
focal_method.return: void
focal_method.signature: void initializeForJob(final Environment env,
  		String operatorIdentifier,
  		TypeSerializer<?> keySerializer)