@Override
public void close(long timeout, TimeUnit unit) {
	synchronized (producerClosingLock) {
		kafkaProducer.close(timeout, unit);
		if (LOG.isDebugEnabled()) {
			LOG.debug(
					"Closed internal KafkaProducer {}. Stacktrace: {}",
					System.identityHashCode(this),
					Joiner.on("\n").join(Thread.currentThread().getStackTrace()));
		}
		closed = true;
	}
}
 

@Override
public void close(Duration duration) {
	synchronized (producerClosingLock) {
		kafkaProducer.close(duration);
		if (LOG.isDebugEnabled()) {
			LOG.debug(
					"Closed internal KafkaProducer {}. Stacktrace: {}",
					System.identityHashCode(this),
					Joiner.on("\n").join(Thread.currentThread().getStackTrace()));
		}
		closed = true;
	}
}
 

@Override
public void close() {
	synchronized (producerClosingLock) {
		kafkaProducer.close();
		if (LOG.isDebugEnabled()) {
			LOG.debug(
					"Closed internal KafkaProducer {}. Stacktrace: {}",
					System.identityHashCode(this),
					Joiner.on("\n").join(Thread.currentThread().getStackTrace()));
		}
		closed = true;
	}
}
 

@Override
public void close(long timeout, TimeUnit unit) {
	synchronized (producerClosingLock) {
		kafkaProducer.close(timeout, unit);
		if (LOG.isDebugEnabled()) {
			LOG.debug(
					"Closed internal KafkaProducer {}. Stacktrace: {}",
					System.identityHashCode(this),
					Joiner.on("\n").join(Thread.currentThread().getStackTrace()));
		}
		closed = true;
	}
}
 

@Override
public void apply(String k, TimeWindow window, Iterable<Tuple2<String, Integer>> input, Collector<String> out) throws Exception {
	out.collect("GOT: " + Joiner.on(",").join(input));
}
 

@Override
public void apply(String k, TimeWindow window, Iterable<Tuple2<String, Integer>> input, Collector<String> out) throws Exception {
	out.collect("GOT: " + Joiner.on(",").join(input));
}
 

@Override
public void apply(String k, TimeWindow window, Iterable<Tuple2<String, Integer>> input, Collector<String> out) throws Exception {
	out.collect("GOT: " + Joiner.on(",").join(input));
}
 

/**
 * Convert column name array to SQL clause.
 *
 * <p>For example, columns "{a, b, c}" will be converted into a SQL-compatible select string section: "`a`, `b`, `c`".
 *
 * @param colNames columns to convert
 * @return converted SQL clause.
 */
public static String columnsToSqlClause(String[] colNames) {
	return Joiner.on("`,`").appendTo(new StringBuilder("`"), colNames).append("`").toString();
}