Skip to content

KAFKA-7551:Refactor to create producer & consumer in the worker. #5842

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
mageshn wants to merge 6 commits into from
Closed

KAFKA-7551:Refactor to create producer & consumer in the worker. #5842

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
79e14c3
Worker Task refactoring
mageshn Oct 22, 2018
de7adec
Remove unused imports
mageshn Oct 25, 2018
57dc817
Fix review comments
mageshn Nov 1, 2018
0467de3
Fix review comments
mageshn Nov 28, 2018
9922269
Fix build failures
mageshn Nov 28, 2018
9124ecf
Fix test failures
mageshn Nov 28, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 46 additions & 16 deletions connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,8 @@
*/
package org.apache.kafka.connect.runtime;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.MetricName;
Expand Down Expand Up @@ -50,6 +52,7 @@
import org.apache.kafka.connect.storage.OffsetStorageReaderImpl;
import org.apache.kafka.connect.storage.OffsetStorageWriter;
import org.apache.kafka.connect.util.ConnectorTaskId;
import org.apache.kafka.connect.util.SinkUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand Down Expand Up @@ -89,7 +92,6 @@ public class Worker {
private final Converter internalKeyConverter;
private final Converter internalValueConverter;
private final OffsetBackingStore offsetBackingStore;
private final Map<String, Object> producerProps;

private final ConcurrentMap<String, WorkerConnector> connectors = new ConcurrentHashMap<>();
private final ConcurrentMap<ConnectorTaskId, WorkerTask> tasks = new ConcurrentHashMap<>();
Expand Down Expand Up @@ -129,19 +131,6 @@ public Worker(

this.workerConfigTransformer = initConfigTransformer();

producerProps = new HashMap<>();
producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, Utils.join(config.getList(WorkerConfig.BOOTSTRAP_SERVERS_CONFIG), ","));
producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
// These settings are designed to ensure there is no data loss. They *may* be overridden via configs passed to the
// worker, but this may compromise the delivery guarantees of Kafka Connect.
producerProps.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, Integer.toString(Integer.MAX_VALUE));
producerProps.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, Long.toString(Long.MAX_VALUE));
producerProps.put(ProducerConfig.ACKS_CONFIG, "all");
producerProps.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "1");
producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, Integer.toString(Integer.MAX_VALUE));
// User-specified overrides
producerProps.putAll(config.originalsWithPrefix("producer."));
}

private WorkerConfigTransformer initConfigTransformer() {
Expand Down Expand Up @@ -498,6 +487,7 @@ private WorkerTask buildWorkerTask(ClusterConfigState configState,
internalKeyConverter, internalValueConverter);
OffsetStorageWriter offsetWriter = new OffsetStorageWriter(offsetBackingStore, id.connector(),
internalKeyConverter, internalValueConverter);
Map<String, Object> producerProps = producerConfigs(config);
KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(producerProps);

// Note we pass the configState as it performs dynamic transformations under the covers
Expand All @@ -508,15 +498,54 @@ private WorkerTask buildWorkerTask(ClusterConfigState configState,
TransformationChain<SinkRecord> transformationChain = new TransformationChain<>(connConfig.<SinkRecord>transformations(), retryWithToleranceOperator);
SinkConnectorConfig sinkConfig = new SinkConnectorConfig(plugins, connConfig.originalsStrings());
retryWithToleranceOperator.reporters(sinkTaskReporters(id, sinkConfig, errorHandlingMetrics));

Map<String, Object> consumerProps = consumerConfigs(id, config);
KafkaConsumer<byte[], byte[]> consumer = new KafkaConsumer<>(consumerProps);

return new WorkerSinkTask(id, (SinkTask) task, statusListener, initialState, config, configState, metrics, keyConverter,
valueConverter, headerConverter, transformationChain, loader, time,
retryWithToleranceOperator);
valueConverter, headerConverter, transformationChain, consumer, loader, time,
retryWithToleranceOperator);
} else {
log.error("Tasks must be a subclass of either SourceTask or SinkTask", task);
throw new ConnectException("Tasks must be a subclass of either SourceTask or SinkTask");
}
}

static Map<String, Object> producerConfigs(WorkerConfig config) {
Map<String, Object> producerProps = new HashMap<>();
producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, Utils.join(config.getList(WorkerConfig.BOOTSTRAP_SERVERS_CONFIG), ","));
producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
// These settings are designed to ensure there is no data loss. They *may* be overridden via configs passed to the
// worker, but this may compromise the delivery guarantees of Kafka Connect.
Comment thread
mageshn marked this conversation as resolved.
Outdated
producerProps.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, Integer.toString(Integer.MAX_VALUE));
producerProps.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, Long.toString(Long.MAX_VALUE));
producerProps.put(ProducerConfig.ACKS_CONFIG, "all");
producerProps.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION, "1");
producerProps.put(ProducerConfig.DELIVERY_TIMEOUT_MS_CONFIG, Integer.toString(Integer.MAX_VALUE));
// User-specified overrides
producerProps.putAll(config.originalsWithPrefix("producer."));
return producerProps;
}


static Map<String, Object> consumerConfigs(ConnectorTaskId id, WorkerConfig config) {
// Include any unknown worker configs so consumer configs can be set globally on the worker
// and through to the task
Comment thread
mageshn marked this conversation as resolved.
Outdated
Map<String, Object> consumerProps = new HashMap<>();

consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, SinkUtils.consumerGroupId(id.connector()));
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
Utils.join(config.getList(WorkerConfig.BOOTSTRAP_SERVERS_CONFIG), ","));
consumerProps.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");

consumerProps.putAll(config.originalsWithPrefix("consumer."));
return consumerProps;
}

ErrorHandlingMetrics errorHandlingMetrics(ConnectorTaskId id) {
return new ErrorHandlingMetrics(id, metrics);
}
Expand All @@ -530,6 +559,7 @@ private List<ErrorReporter> sinkTaskReporters(ConnectorTaskId id, SinkConnectorC
// check if topic for dead letter queue exists
String topic = connConfig.dlqTopicName();
if (topic != null && !topic.isEmpty()) {
Map<String, Object> producerProps = producerConfigs(config);
DeadLetterQueueReporter reporter = DeadLetterQueueReporter.createAndSetup(config, id, connConfig, producerProps, errorHandlingMetrics);
reporters.add(reporter);
}
Expand Down
31 changes: 2 additions & 29 deletions connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerSinkTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@
*/
package org.apache.kafka.connect.runtime;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
Expand All @@ -33,7 +32,6 @@
import org.apache.kafka.common.metrics.stats.Total;
import org.apache.kafka.common.metrics.stats.Value;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.common.utils.Utils;
import org.apache.kafka.connect.data.SchemaAndValue;
import org.apache.kafka.connect.errors.ConnectException;
import org.apache.kafka.connect.errors.RetriableException;
Expand All @@ -49,7 +47,6 @@
import org.apache.kafka.connect.storage.HeaderConverter;
import org.apache.kafka.connect.util.ConnectUtils;
import org.apache.kafka.connect.util.ConnectorTaskId;
import org.apache.kafka.connect.util.SinkUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand Down Expand Up @@ -105,6 +102,7 @@ public WorkerSinkTask(ConnectorTaskId id,
Converter valueConverter,
HeaderConverter headerConverter,
TransformationChain<SinkRecord> transformationChain,
KafkaConsumer<byte[], byte[]> consumer,
ClassLoader loader,
Time time,
RetryWithToleranceOperator retryWithToleranceOperator) {
Expand All @@ -131,13 +129,13 @@ public WorkerSinkTask(ConnectorTaskId id,
this.commitFailures = 0;
this.sinkTaskMetricsGroup = new SinkTaskMetricsGroup(id, connectMetrics);
this.sinkTaskMetricsGroup.recordOffsetSequenceNumber(commitSeqno);
this.consumer = consumer;
}

@Override
public void initialize(TaskConfig taskConfig) {
try {
this.taskConfig = taskConfig.originalsStrings();
this.consumer = createConsumer();
this.context = new WorkerSinkTaskContext(consumer, this, configState);
} catch (Throwable t) {
log.error("{} Task failed initialization and will not be started.", this, t);
Expand Down Expand Up @@ -455,31 +453,6 @@ private ConsumerRecords<byte[], byte[]> pollConsumer(long timeoutMs) {
return msgs;
}

private KafkaConsumer<byte[], byte[]> createConsumer() {
// Include any unknown worker configs so consumer configs can be set globally on the worker
// and through to the task
Map<String, Object> props = new HashMap<>();

props.put(ConsumerConfig.GROUP_ID_CONFIG, SinkUtils.consumerGroupId(id.connector()));
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
Utils.join(workerConfig.getList(WorkerConfig.BOOTSTRAP_SERVERS_CONFIG), ","));
props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");

props.putAll(workerConfig.originalsWithPrefix("consumer."));

KafkaConsumer<byte[], byte[]> newConsumer;
try {
newConsumer = new KafkaConsumer<>(props);
} catch (Throwable t) {
throw new ConnectException("Failed to create consumer", t);
}

return newConsumer;
}

private void convertMessages(ConsumerRecords<byte[], byte[]> msgs) {
origOffsets.clear();
for (ConsumerRecord<byte[], byte[]> msg : msgs) {
Expand Down
10 changes: 4 additions & 6 deletions connect/runtime/src/test/java/org/apache/kafka/connect/runtime/ErrorHandlingTaskTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -352,7 +352,6 @@ private void assertErrorHandlingMetricValue(String name, double expected) {
}

private void expectInitializeTask() throws Exception {
PowerMock.expectPrivate(workerSinkTask, "createConsumer").andReturn(consumer);
consumer.subscribe(EasyMock.eq(singletonList(TOPIC)), EasyMock.capture(rebalanceListener));
PowerMock.expectLastCall();

Expand All @@ -371,11 +370,10 @@ private void createSinkTask(TargetState initialState, RetryWithToleranceOperator

TransformationChain<SinkRecord> sinkTransforms = new TransformationChain<>(singletonList(new FaultyPassthrough<SinkRecord>()), retryWithToleranceOperator);

workerSinkTask = PowerMock.createPartialMock(
WorkerSinkTask.class, new String[]{"createConsumer"},
taskId, sinkTask, statusListener, initialState, workerConfig,
ClusterConfigState.EMPTY, metrics, converter, converter,
headerConverter, sinkTransforms, pluginLoader, time, retryWithToleranceOperator);
workerSinkTask = new WorkerSinkTask(
taskId, sinkTask, statusListener, initialState, workerConfig,
ClusterConfigState.EMPTY, metrics, converter, converter,
headerConverter, sinkTransforms, consumer, pluginLoader, time, retryWithToleranceOperator);
}

private void createSourceTask(TargetState initialState, RetryWithToleranceOperator retryWithToleranceOperator) {
Expand Down
13 changes: 5 additions & 8 deletions connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -162,12 +162,11 @@ public void setUp() {
}

private void createTask(TargetState initialState) {
workerTask = PowerMock.createPartialMock(
WorkerSinkTask.class, new String[]{"createConsumer"},
taskId, sinkTask, statusListener, initialState, workerConfig, ClusterConfigState.EMPTY, metrics,
keyConverter, valueConverter, headerConverter,
transformationChain, pluginLoader, time,
RetryWithToleranceOperatorTest.NOOP_OPERATOR);
workerTask = new WorkerSinkTask(

@ewencp ewencp Nov 27, 2018

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I can't remember all the details of when the powermock runner ends up being required, but this looks like we may have removed any need for powermock in this test other than Whitebox which may not need it. Would be worth figuring out if we can clean out some powermock usage since it's mostly only used in Connect and has caused problems with newer JDKs.

We don't need to do this here, but if we clean up uses of PowerMock we should consider doing cleanup passes for things like the test runner. Alternatively, might make sense to, at some point, do a JIRA dedicated to just removing use of powermock.

@mageshn mageshn Nov 29, 2018

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I have created a test to address this for Connect. I haven't looked upto see if other components are using it as well. If they do, I will create tickets for those as well. https://issues.apache.org/jira/browse/KAFKA-7686

taskId, sinkTask, statusListener, initialState, workerConfig, ClusterConfigState.EMPTY, metrics,
keyConverter, valueConverter, headerConverter,
transformationChain, consumer, pluginLoader, time,
RetryWithToleranceOperatorTest.NOOP_OPERATOR);
Comment thread
mageshn marked this conversation as resolved.
Outdated
}

@After
Expand Down Expand Up @@ -1167,7 +1166,6 @@ public void testTopicsRegex() throws Exception {

createTask(TargetState.PAUSED);

PowerMock.expectPrivate(workerTask, "createConsumer").andReturn(consumer);
consumer.subscribe(EasyMock.capture(topicsRegex), EasyMock.capture(rebalanceListener));
PowerMock.expectLastCall();

Expand Down Expand Up @@ -1255,7 +1253,6 @@ public void testMetricsGroup() {
}

private void expectInitializeTask() throws Exception {
PowerMock.expectPrivate(workerTask, "createConsumer").andReturn(consumer);
consumer.subscribe(EasyMock.eq(asList(TOPIC)), EasyMock.capture(rebalanceListener));
PowerMock.expectLastCall();

Expand Down
6 changes: 2 additions & 4 deletions ...ct/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerSinkTaskThreadedTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,12 +137,11 @@ public void setup() {
workerProps.put("offset.storage.file.filename", "/tmp/connect.offsets");
pluginLoader = PowerMock.createMock(PluginClassLoader.class);
workerConfig = new StandaloneConfig(workerProps);
workerTask = PowerMock.createPartialMock(
WorkerSinkTask.class, new String[]{"createConsumer"},
workerTask = new WorkerSinkTask(
taskId, sinkTask, statusListener, initialState, workerConfig, ClusterConfigState.EMPTY, metrics, keyConverter,
valueConverter, headerConverter,
new TransformationChain<>(Collections.emptyList(), RetryWithToleranceOperatorTest.NOOP_OPERATOR),
pluginLoader, time, RetryWithToleranceOperatorTest.NOOP_OPERATOR);
consumer, pluginLoader, time, RetryWithToleranceOperatorTest.NOOP_OPERATOR);

recordsReturned = 0;
}
Expand Down Expand Up @@ -509,7 +508,6 @@ public Object answer() throws Throwable {
}

private void expectInitializeTask() throws Exception {
PowerMock.expectPrivate(workerTask, "createConsumer").andReturn(consumer);

consumer.subscribe(EasyMock.eq(Arrays.asList(TOPIC)), EasyMock.capture(rebalanceListener));
PowerMock.expectLastCall();
Expand Down
Loading