Victor Osório vepo

Kafka 0.11.0.0 (Confluent 3.3.0) added support to manipulate offsets for a consumer group via cli kafka-consumer-groups command.

List the topics to which the group is subscribed

kafka-consumer-groups --bootstrap-server <kafkahost:port> --group <group_id> --describe

Note the values under "CURRENT-OFFSET" and "LOG-END-OFFSET". "CURRENT-OFFSET" is the offset where this consumer group is currently at in each of the partitions.

Reset the consumer offset for a topic (preview)

	URL[] jars = new URL[] { /* The jars to be loaded */ }; //
	try (URLClassLoader classLoader = new URLClassLoader(jars)) {
	Job job = classLoader.loadClass(Job.class.getName());
	job.execute();
	}

	URL[] jars = new URL[] { /* The jars to be loaded */ }; //
	try (URLClassLoader classLoader = new URLClassLoader(jars)) {
	Job job = classLoader.loadClass(Job.class.getName());
	job.execute();
	Collections.list(DriverManager.getDrivers()).forEach(driver -> {
	try {
	DriverManager.deregisterDriver(driver);
	} catch (SQLException e) {
	logger.error("Error unregistering driver!", e);
	}

	<configuration>
	<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
	<encoder>
	<pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
	</encoder>
	</appender>

	<appender name="FILE" class="ch.qos.logback.core.FileAppender">
	<file>execution.log</file>
	<encoder>

	@Test
	@DisplayName("It should fail for invalid URL")
	public void couldNotConnectTest() {
	// Prepare the test
	InetSocketAddress remoteAddress = remoteAddress();
	validateHttp("/defaultGet", "GET", 200, "{\"response\":\"OK\"}", 0, () -> {
	String port = Integer.toString(remoteAddress.getPort());
	Suite suite = parseSuite(HTTP_GET_TEST_SUITE.replace("${host}", "not-a-valid-endpoint.com.br")
	.replace("${port}", port));

	private Result executeStep(Step step, Context context) {
	// The plugin exists?
	if (stepExecutors.containsKey(step.getPlugin())) {
	StepExecutor executor = stepExecutors.get(step.getPlugin());
	// Check if there is any missing attribute
	Set<Attribute<?>> missingAttributes = executor.requiredAttribute()
	.filter(Attribute::isRequired)
	.filter(entry -> !step.getAttributes()
	.containsKey(entry.getKey()))
	.collect(toSet());

	private Result executeStep(Step step, Context context) {
	return checkAssertions(step, stepExecutors.get(step.getPlugin()).execute(step, context));
	}

	package io.vepo.kafka.articles;

	import static java.util.Arrays.asList;

	import java.time.Duration;
	import java.util.Properties;

	import org.apache.kafka.clients.consumer.Consumer;
	import org.apache.kafka.clients.consumer.ConsumerConfig;
	import org.apache.kafka.clients.consumer.ConsumerRecord;

	package io.vepo.kafka.articles;

	import java.util.Properties;
	import java.util.Scanner;
	import java.util.concurrent.Future;

	import org.apache.kafka.clients.producer.KafkaProducer;
	import org.apache.kafka.clients.producer.Producer;
	import org.apache.kafka.clients.producer.ProducerConfig;
	import org.apache.kafka.clients.producer.ProducerRecord;

	package io.vepo.access.user;

	import static com.mongodb.client.model.Filters.and;
	import static com.mongodb.client.model.Filters.eq;

	import java.util.ArrayList;
	import java.util.List;
	import java.util.Optional;

	import javax.enterprise.context.Dependent;