You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2020/05/24 14:46:56 UTC
[GitHub] [kafka] kkonstantine commented on a change in pull request #2604: KAFKA-4794: Add access to OffsetStorageReader from SourceConnector
kkonstantine commented on a change in pull request #2604:
URL: https://github.com/apache/kafka/pull/2604#discussion_r429641860
##########
File path: connect/api/src/test/java/org/apache/kafka/connect/connector/ConnectorTest.java
##########
@@ -0,0 +1,198 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.connect.connector;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.kafka.common.config.ConfigDef;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+public class ConnectorTest {
+
+ protected ConnectorContext context;
+ protected Connector connector;
+ protected AssertableConnector assertableConnector;
+
+ @Before
+ public void beforeEach() {
+ connector = createConnector();
+ context = createContext();
+ assertableConnector = (AssertableConnector) connector;
+ }
+
+ @Test
+ public void shouldInitializeContext() {
+ connector.initialize(context);
+ assertableConnector.assertInitialized();
+ assertableConnector.assertContext(context);
+ assertableConnector.assertTaskConfigs(null);
+ }
+
+ @Test
+ public void shouldInitializeContextWithTaskConfigs() {
+ List<Map<String, String>> taskConfigs = new ArrayList<>();
+ connector.initialize(context, taskConfigs);
+ assertableConnector.assertInitialized();
+ assertableConnector.assertContext(context);
+ assertableConnector.assertTaskConfigs(taskConfigs);
+ }
+
+ @Test
+ public void shouldStopAndStartWhenReconfigure() {
+ Map<String, String> props = new HashMap<>();
+ connector.initialize(context);
+ assertableConnector.assertContext(context);
+ assertableConnector.assertStarted(false);
+ assertableConnector.assertStopped(false);
+ connector.reconfigure(props);
+ assertableConnector.assertStarted(true);
+ assertableConnector.assertStopped(true);
+ assertableConnector.assertProperties(props);
+ }
+
+ protected ConnectorContext createContext() {
+ return new TestConnectorContext();
+ }
+
+ protected Connector createConnector() {
+ return new TestConnector();
+ }
+
+ public static class TestConnectorContext implements ConnectorContext {
+
+ @Override
+ public void requestTaskReconfiguration() {
+ }
+
+ @Override
+ public void raiseError(Exception e) {
+ }
+ }
+
+ public interface AssertableConnector {
+
+ void assertContext(ConnectorContext expected);
+
+ void assertInitialized();
+
+ void assertTaskConfigs(List<Map<String, String>> expectedTaskConfigs);
+
+ void assertStarted(boolean expected);
+
+ void assertStopped(boolean expected);
+
+ void assertProperties(Map<String, String> expected);
+
+ }
+
+ public static class TestConnector extends Connector implements AssertableConnector {
Review comment:
Should we remove this class? Currently they can only be 2 types of connectors, source or sink connectors. This seems to introduce a third implementation of the `abstract Connector` class, but testing seems sufficient without it.
##########
File path: connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConnector.java
##########
@@ -58,18 +61,21 @@
private Map<String, String> config;
private State state;
-
- public WorkerConnector(String connName,
- Connector connector,
- ConnectorContext ctx,
- ConnectMetrics metrics,
- ConnectorStatus.Listener statusListener) {
+ private final OffsetStorageReader offsetStorageReader;
+
+ public WorkerConnector(final String connName,
Review comment:
adding `final` in every argument here was just noise but with not much added value. Given that we don't write constructors like this in Connect I think we should revert and add the new argument without the `final` specifier.
##########
File path: connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConnector.java
##########
@@ -319,4 +320,36 @@ protected MetricGroup metricGroup() {
return metricGroup;
}
}
+
+ private class WorkerConnectorContext implements ConnectorContext {
+
+ @Override
+ public void requestTaskReconfiguration() {
+ WorkerConnector.this.ctx.requestTaskReconfiguration();
+ }
+
+ @Override
+ public void raiseError(Exception e) {
+ log.error("{} Connector raised an error", WorkerConnector.this, e);
+ onFailure(e);
+ WorkerConnector.this.ctx.raiseError(e);
+ }
+ }
+
+ private class WorkerSinkConnectorContext extends WorkerConnectorContext implements SinkConnectorContext {
+ }
+
+ private class WorkerSourceConnectorContext extends WorkerConnectorContext implements SourceConnectorContext {
+
+ private final OffsetStorageReader offsetStorageReader;
+
+ WorkerSourceConnectorContext(final OffsetStorageReader offsetStorageReader) {
Review comment:
```suggestion
WorkerSourceConnectorContext(OffsetStorageReader offsetStorageReader) {
```
##########
File path: connect/runtime/src/main/java/org/apache/kafka/connect/runtime/WorkerConnector.java
##########
@@ -80,19 +86,14 @@ public void initialize(ConnectorConfig connectorConfig) {
SinkConnectorConfig.validate(config);
}
- connector.initialize(new ConnectorContext() {
- @Override
- public void requestTaskReconfiguration() {
- ctx.requestTaskReconfiguration();
- }
-
- @Override
- public void raiseError(Exception e) {
- log.error("{} Connector raised an error", WorkerConnector.this, e);
- onFailure(e);
- ctx.raiseError(e);
- }
- });
+ if (isSinkConnector()) {
+ SinkConnectorConfig.validate(config);
+ connector.initialize(new WorkerSinkConnectorContext());
+ } else if (isSourceConnector()) {
+ connector.initialize(new WorkerSourceConnectorContext(offsetStorageReader));
+ } else {
+ connector.initialize(new WorkerConnectorContext());
Review comment:
We don't allow a third option [elsewhere](https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java#L1288-L1292) in the code, so probably we shouldn't here either.
##########
File path: connect/api/src/main/java/org/apache/kafka/connect/source/SourceConnector.java
##########
@@ -24,4 +24,11 @@
*/
public abstract class SourceConnector extends Connector {
+ /**
+ * {@inheritDoc}
+ */
Review comment:
```suggestion
```
we are not adding anything to the javadoc of the super method, so we shouldn't need `@inheritDoc`
##########
File path: connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerConnectorTest.java
##########
@@ -416,6 +436,48 @@ public void testTransitionPausedToPaused() {
verifyAll();
}
+ @Test
+ public void testInitializeConnectorThatIsNeitherSourceNorSink() {
Review comment:
I'm not convinced we should allow this
##########
File path: connect/api/src/main/java/org/apache/kafka/connect/sink/SinkConnector.java
##########
@@ -34,4 +34,12 @@
*/
public static final String TOPICS_CONFIG = "topics";
+ /**
+ * {@inheritDoc}
+ */
Review comment:
```suggestion
```
we are not adding anything to the javadoc of the super method, so we shouldn't need `@inheritDoc`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org