You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "C0urante (via GitHub)" <gi...@apache.org> on 2023/12/07 20:47:03 UTC
[PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
C0urante opened a new pull request, #14966:
URL: https://github.com/apache/kafka/pull/14966
[Jira](https://issues.apache.org/jira/browse/KAFKA-15988)
Using a similar approach to the existing [ConnectorRestartApiIntegrationTest suite](https://github.com/apache/kafka/blob/c062e5a1f91f44c721072a29b533304d58c487ab/connect/runtime/src/test/java/org/apache/kafka/connect/integration/ConnectorRestartApiIntegrationTest.java#L83-L129), embedded Kafka clusters and embedded Connect clusters that use the same worker configs are reused across test cases.
In order to reduce possible interference across test cases, the name of the connector and the topic it interacts with for each case is derived from the name of the test case. In the same vein, overridden sink connector group IDs and source connector offsets topics are also derived from the name of the test case.
Although the diff is a few hundred lines long, it mostly boils down to these simple changes:
- Replace the `CONNECTOR_NAME` constant with the `connectorName` field
- Replace the `TOPIC` constant with the `topic` field
- Replace hardcoded overridden sink connector consumer group IDs and source connector offsets topics with names that now include the test case
- Add an internal API to reuse `EmbeddedConnectCluster` instances (based on their worker configs) across tests, which also reuses their underlying `EmbeddedKafkaCluster` instances
- Add cleanup logic at the end of each test to delete the test-specific connector, and fail the test if any other leaked connectors are found
The last point does mean that we may double the number of concurrently-running embedded Kafka and Connect clusters. With local testing, this was worth the tradeoff: test runtime decreased from 6 minutes and 17 seconds to 3 minutes and 27 seconds. However, we may explore an improvement in the future where we order tests by the kind of embedded Connect cluster they require, and bring down any currently-running embedded Connect cluster before bringing up a new one.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on PR #14966:
URL: https://github.com/apache/kafka/pull/14966#issuecomment-1852273484
Gradle Enterprise shows an over 75% reduction in the runtime for this test suite.
[Test runs from this PR](https://ge.apache.org/scans/tests?search.tags=PR-14966&tests.container=org.apache.kafka.connect.integration.OffsetsApiIntegrationTest&tests.unstableOnly=false) average 5 minutes and 22 seconds, while [test runs on trunk](https://ge.apache.org/scans/tests?search.tags=trunk&tests.container=org.apache.kafka.connect.integration.OffsetsApiIntegrationTest&tests.unstableOnly=false) average 24 minutes and 22 seconds.
Given that, I believe this is ready for review.
@gharris1727 @yashmayya would either of you have time to take a look?
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #14966:
URL: https://github.com/apache/kafka/pull/14966#discussion_r1420935251
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -709,104 +756,124 @@ public void testResetSinkConnectorOffsetsDifferentKafkaClusterTargeted() throws
private void resetAndVerifySinkConnectorOffsets(Map<String, String> connectorConfigs, EmbeddedKafkaCluster kafkaCluster) throws Exception {
int numPartitions = 3;
- kafkaCluster.createTopic(TOPIC, numPartitions);
+ kafkaCluster.createTopic(topic, numPartitions);
// Produce records to each partition
for (int partition = 0; partition < numPartitions; partition++) {
for (int record = 0; record < NUM_RECORDS_PER_PARTITION; record++) {
- kafkaCluster.produce(TOPIC, partition, "key", "value");
+ kafkaCluster.produce(topic, partition, "key", "value");
}
}
// Create sink connector
- connect.configureConnector(CONNECTOR_NAME, connectorConfigs);
- connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(CONNECTOR_NAME, NUM_TASKS,
+ connect.configureConnector(connectorName, connectorConfigs);
+ connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(connectorName, NUM_TASKS,
"Connector tasks did not start in time.");
- verifyExpectedSinkConnectorOffsets(CONNECTOR_NAME, TOPIC, numPartitions, NUM_RECORDS_PER_PARTITION,
+ verifyExpectedSinkConnectorOffsets(connectorName, topic, numPartitions, NUM_RECORDS_PER_PARTITION,
"Sink connector consumer group offsets should catch up to the topic end offsets");
- connect.stopConnector(CONNECTOR_NAME);
+ connect.stopConnector(connectorName);
connect.assertions().assertConnectorIsStopped(
- CONNECTOR_NAME,
+ connectorName,
"Connector did not stop in time"
);
+ final AtomicReference<String> responseReference = new AtomicReference<>();
// Reset the sink connector's offsets
- String response = connect.resetConnectorOffsets(CONNECTOR_NAME);
+ // Some retry logic is necessary to account for KAFKA-15826,
+ // where laggy sink task startup/shutdown can leave consumers running
+ waitForCondition(
+ () -> {
+ try {
+ responseReference.set(connect.resetConnectorOffsets(connectorName));
+ return true;
+ } catch (ConnectRestException e) {
+ boolean internalServerError = e.statusCode() == INTERNAL_SERVER_ERROR.getStatusCode();
+
+ String message = Optional.of(e.getMessage()).orElse("");
+ boolean failedToResetConsumerOffsets = message.contains("Failed to reset consumer group offsets for connector");
+ boolean canBeRetried = message.contains("If the connector is in a stopped state, this operation can be safely retried");
+
+ boolean retriable = internalServerError && failedToResetConsumerOffsets && canBeRetried;
+ if (retriable) {
+ return false;
+ } else {
+ throw new NoRetryException(e);
+ }
+ } catch (Throwable t) {
+ throw new NoRetryException(t);
+ }
+ },
+ 30_000,
+ "Failed to reset sink connector offsets in time"
+ );
+ String response = responseReference.get();
Review Comment:
Same as above: this part isn't related to optimizing test runtime, it's just meant to tame some of the flakiness I've seen in CI builds for this PR.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #14966:
URL: https://github.com/apache/kafka/pull/14966#discussion_r1420935041
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -324,127 +363,151 @@ public void testAlterSinkConnectorOffsetsDifferentKafkaClusterTargeted() throws
private void alterAndVerifySinkConnectorOffsets(Map<String, String> connectorConfigs, EmbeddedKafkaCluster kafkaCluster) throws Exception {
int numPartitions = 3;
- kafkaCluster.createTopic(TOPIC, numPartitions);
+ kafkaCluster.createTopic(topic, numPartitions);
// Produce records to each partition
for (int partition = 0; partition < numPartitions; partition++) {
for (int record = 0; record < NUM_RECORDS_PER_PARTITION; record++) {
- kafkaCluster.produce(TOPIC, partition, "key", "value");
+ kafkaCluster.produce(topic, partition, "key", "value");
}
}
// Create sink connector
- connect.configureConnector(CONNECTOR_NAME, connectorConfigs);
- connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(CONNECTOR_NAME, NUM_TASKS,
+ connect.configureConnector(connectorName, connectorConfigs);
+ connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(connectorName, NUM_TASKS,
"Connector tasks did not start in time.");
- verifyExpectedSinkConnectorOffsets(CONNECTOR_NAME, TOPIC, numPartitions, NUM_RECORDS_PER_PARTITION,
+ verifyExpectedSinkConnectorOffsets(connectorName, topic, numPartitions, NUM_RECORDS_PER_PARTITION,
"Sink connector consumer group offsets should catch up to the topic end offsets");
- connect.stopConnector(CONNECTOR_NAME);
+ connect.stopConnector(connectorName);
connect.assertions().assertConnectorIsStopped(
- CONNECTOR_NAME,
+ connectorName,
"Connector did not stop in time"
);
// Delete the offset of one partition; alter the offsets of the others
List<ConnectorOffset> offsetsToAlter = new ArrayList<>();
Map<String, Object> partition = new HashMap<>();
- partition.put(SinkUtils.KAFKA_TOPIC_KEY, TOPIC);
+ partition.put(SinkUtils.KAFKA_TOPIC_KEY, topic);
partition.put(SinkUtils.KAFKA_PARTITION_KEY, 0);
offsetsToAlter.add(new ConnectorOffset(partition, null));
for (int i = 1; i < numPartitions; i++) {
partition = new HashMap<>();
- partition.put(SinkUtils.KAFKA_TOPIC_KEY, TOPIC);
+ partition.put(SinkUtils.KAFKA_TOPIC_KEY, topic);
partition.put(SinkUtils.KAFKA_PARTITION_KEY, i);
offsetsToAlter.add(new ConnectorOffset(partition, Collections.singletonMap(SinkUtils.KAFKA_OFFSET_KEY, 5)));
}
- String response = connect.alterConnectorOffsets(CONNECTOR_NAME, new ConnectorOffsets(offsetsToAlter));
+ AtomicReference<String> responseReference = new AtomicReference<>();
+ // Some retry logic is necessary to account for KAFKA-15826,
+ // where laggy sink task startup/shutdown can leave consumers running
+ waitForCondition(
+ () -> {
+ try {
+ responseReference.set(connect.alterConnectorOffsets(connectorName, new ConnectorOffsets(offsetsToAlter)));
+ return true;
+ } catch (ConnectRestException e) {
+ boolean internalServerError = e.statusCode() == INTERNAL_SERVER_ERROR.getStatusCode();
+
+ String message = Optional.of(e.getMessage()).orElse("");
+ boolean failedToResetConsumerOffsets = message.contains("Failed to reset consumer group offsets for connector");
+ boolean canBeRetried = message.contains("If the connector is in a stopped state, this operation can be safely retried");
+
+ boolean retriable = internalServerError && failedToResetConsumerOffsets && canBeRetried;
+ if (retriable) {
+ return false;
+ } else {
+ throw new NoRetryException(e);
+ }
+ } catch (Throwable t) {
+ throw new NoRetryException(t);
+ }
+ },
+ 30_000,
+ "Failed to reset sink connector offsets in time"
+ );
+ String response = responseReference.get();
Review Comment:
This part isn't related to optimizing test runtime, it's just meant to tame some of the flakiness I've seen in CI builds for this PR.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #14966:
URL: https://github.com/apache/kafka/pull/14966#discussion_r1425552879
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -70,74 +80,115 @@ public class OffsetsApiIntegrationTest {
private static final long OFFSET_COMMIT_INTERVAL_MS = TimeUnit.SECONDS.toMillis(1);
private static final long OFFSET_READ_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(30);
private static final int NUM_WORKERS = 3;
- private static final String CONNECTOR_NAME = "test-connector";
- private static final String TOPIC = "test-topic";
private static final int NUM_TASKS = 2;
private static final int NUM_RECORDS_PER_PARTITION = 10;
- private Map<String, String> workerProps;
- private EmbeddedConnectCluster.Builder connectBuilder;
+ private static final Map<Map<String, String>, EmbeddedConnectCluster> CONNECT_CLUSTERS = new ConcurrentHashMap<>();
+ @Rule
+ public TestName currentTest = new TestName();
private EmbeddedConnectCluster connect;
+ private String connectorName;
+ private String topic;
@Before
public void setup() {
- Properties brokerProps = new Properties();
- brokerProps.put("transaction.state.log.replication.factor", "1");
- brokerProps.put("transaction.state.log.min.isr", "1");
-
- // setup Connect worker properties
- workerProps = new HashMap<>();
- workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
-
- // build a Connect cluster backed by Kafka and Zk
- connectBuilder = new EmbeddedConnectCluster.Builder()
- .name("connect-cluster")
- .numWorkers(NUM_WORKERS)
- .brokerProps(brokerProps)
- .workerProps(workerProps);
+ connectorName = currentTest.getMethodName();
+ topic = currentTest.getMethodName();
+ connect = defaultConnectCluster();
}
@After
public void tearDown() {
- connect.stop();
+ Set<String> remainingConnectors = new HashSet<>(connect.connectors());
+ if (remainingConnectors.remove(connectorName)) {
+ connect.deleteConnector(connectorName);
+ }
+ try {
+ assertEquals(
+ "Some connectors were not properly cleaned up after this test",
+ Collections.emptySet(),
+ remainingConnectors
+ );
+ } finally {
+ // Make a last-ditch effort to clean up the leaked connectors
+ // so as not to interfere with other test cases
+ remainingConnectors.forEach(connect::deleteConnector);
+ }
+ }
+
+ @AfterClass
+ public static void close() {
+ // stop all Connect, Kafka and Zk threads.
+ CONNECT_CLUSTERS.values().forEach(EmbeddedConnectCluster::stop);
+ }
+
+ private static EmbeddedConnectCluster createOrReuseConnectWithWorkerProps(Map<String, String> workerProps) {
+ return CONNECT_CLUSTERS.computeIfAbsent(workerProps, props -> {
+ Properties brokerProps = new Properties();
+ brokerProps.put("transaction.state.log.replication.factor", "1");
+ brokerProps.put("transaction.state.log.min.isr", "1");
+
+ // Have to declare a new map since the passed-in one may be immutable
+ Map<String, String> workerPropsWithDefaults = new HashMap<>(workerProps);
+ // Enable fast offset commits by default
+ workerPropsWithDefaults.putIfAbsent(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
+
+ EmbeddedConnectCluster result = new EmbeddedConnectCluster.Builder()
+ .name("connect-cluster")
+ .numWorkers(NUM_WORKERS)
+ .brokerProps(brokerProps)
+ .workerProps(workerPropsWithDefaults)
+ .build();
+
+ result.start();
+
+ return result;
+ });
+ }
+
+ private static EmbeddedConnectCluster defaultConnectCluster() {
+ return createOrReuseConnectWithWorkerProps(Collections.emptyMap());
+ }
+
+ private static EmbeddedConnectCluster exactlyOnceSourceConnectCluster() {
+ Map<String, String> workerProps = Collections.singletonMap(
+ DistributedConfig.EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG,
+ "enabled"
+ );
+ return createOrReuseConnectWithWorkerProps(workerProps);
}
@Test
public void testGetNonExistentConnectorOffsets() {
- connect = connectBuilder.build();
- connect.start();
ConnectRestException e = assertThrows(ConnectRestException.class,
() -> connect.connectorOffsets("non-existent-connector"));
assertEquals(404, e.errorCode());
}
@Test
public void testGetSinkConnectorOffsets() throws Exception {
- connect = connectBuilder.build();
- connect.start();
getAndVerifySinkConnectorOffsets(baseSinkConnectorConfigs(), connect.kafka());
}
@Test
public void testGetSinkConnectorOffsetsOverriddenConsumerGroupId() throws Exception {
- connect = connectBuilder.build();
- connect.start();
Map<String, String> connectorConfigs = baseSinkConnectorConfigs();
- connectorConfigs.put(ConnectorConfig.CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX + CommonClientConfigs.GROUP_ID_CONFIG,
- "overridden-group-id");
+ String overriddenGroupId = connectorName = "-overridden-group-id";
Review Comment:
😱
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -70,74 +80,115 @@ public class OffsetsApiIntegrationTest {
private static final long OFFSET_COMMIT_INTERVAL_MS = TimeUnit.SECONDS.toMillis(1);
private static final long OFFSET_READ_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(30);
private static final int NUM_WORKERS = 3;
- private static final String CONNECTOR_NAME = "test-connector";
- private static final String TOPIC = "test-topic";
private static final int NUM_TASKS = 2;
private static final int NUM_RECORDS_PER_PARTITION = 10;
- private Map<String, String> workerProps;
- private EmbeddedConnectCluster.Builder connectBuilder;
+ private static final Map<Map<String, String>, EmbeddedConnectCluster> CONNECT_CLUSTERS = new ConcurrentHashMap<>();
+ @Rule
+ public TestName currentTest = new TestName();
private EmbeddedConnectCluster connect;
+ private String connectorName;
+ private String topic;
@Before
public void setup() {
- Properties brokerProps = new Properties();
- brokerProps.put("transaction.state.log.replication.factor", "1");
- brokerProps.put("transaction.state.log.min.isr", "1");
-
- // setup Connect worker properties
- workerProps = new HashMap<>();
- workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
-
- // build a Connect cluster backed by Kafka and Zk
- connectBuilder = new EmbeddedConnectCluster.Builder()
- .name("connect-cluster")
- .numWorkers(NUM_WORKERS)
- .brokerProps(brokerProps)
- .workerProps(workerProps);
+ connectorName = currentTest.getMethodName();
+ topic = currentTest.getMethodName();
+ connect = defaultConnectCluster();
}
@After
public void tearDown() {
- connect.stop();
+ Set<String> remainingConnectors = new HashSet<>(connect.connectors());
+ if (remainingConnectors.remove(connectorName)) {
+ connect.deleteConnector(connectorName);
+ }
+ try {
+ assertEquals(
+ "Some connectors were not properly cleaned up after this test",
+ Collections.emptySet(),
+ remainingConnectors
+ );
+ } finally {
+ // Make a last-ditch effort to clean up the leaked connectors
+ // so as not to interfere with other test cases
+ remainingConnectors.forEach(connect::deleteConnector);
+ }
+ }
+
+ @AfterClass
+ public static void close() {
+ // stop all Connect, Kafka and Zk threads.
+ CONNECT_CLUSTERS.values().forEach(EmbeddedConnectCluster::stop);
+ }
+
+ private static EmbeddedConnectCluster createOrReuseConnectWithWorkerProps(Map<String, String> workerProps) {
+ return CONNECT_CLUSTERS.computeIfAbsent(workerProps, props -> {
+ Properties brokerProps = new Properties();
+ brokerProps.put("transaction.state.log.replication.factor", "1");
+ brokerProps.put("transaction.state.log.min.isr", "1");
+
+ // Have to declare a new map since the passed-in one may be immutable
+ Map<String, String> workerPropsWithDefaults = new HashMap<>(workerProps);
+ // Enable fast offset commits by default
+ workerPropsWithDefaults.putIfAbsent(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
+
+ EmbeddedConnectCluster result = new EmbeddedConnectCluster.Builder()
+ .name("connect-cluster")
+ .numWorkers(NUM_WORKERS)
+ .brokerProps(brokerProps)
+ .workerProps(workerPropsWithDefaults)
+ .build();
+
+ result.start();
+
+ return result;
+ });
+ }
+
+ private static EmbeddedConnectCluster defaultConnectCluster() {
+ return createOrReuseConnectWithWorkerProps(Collections.emptyMap());
+ }
+
+ private static EmbeddedConnectCluster exactlyOnceSourceConnectCluster() {
+ Map<String, String> workerProps = Collections.singletonMap(
+ DistributedConfig.EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG,
+ "enabled"
+ );
+ return createOrReuseConnectWithWorkerProps(workerProps);
}
@Test
public void testGetNonExistentConnectorOffsets() {
- connect = connectBuilder.build();
- connect.start();
ConnectRestException e = assertThrows(ConnectRestException.class,
() -> connect.connectorOffsets("non-existent-connector"));
assertEquals(404, e.errorCode());
}
@Test
public void testGetSinkConnectorOffsets() throws Exception {
- connect = connectBuilder.build();
- connect.start();
getAndVerifySinkConnectorOffsets(baseSinkConnectorConfigs(), connect.kafka());
}
@Test
public void testGetSinkConnectorOffsetsOverriddenConsumerGroupId() throws Exception {
- connect = connectBuilder.build();
- connect.start();
Map<String, String> connectorConfigs = baseSinkConnectorConfigs();
- connectorConfigs.put(ConnectorConfig.CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX + CommonClientConfigs.GROUP_ID_CONFIG,
- "overridden-group-id");
+ String overriddenGroupId = connectorName = "-overridden-group-id";
Review Comment:
😱
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #14966:
URL: https://github.com/apache/kafka/pull/14966#discussion_r1424453814
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -70,74 +80,115 @@ public class OffsetsApiIntegrationTest {
private static final long OFFSET_COMMIT_INTERVAL_MS = TimeUnit.SECONDS.toMillis(1);
private static final long OFFSET_READ_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(30);
private static final int NUM_WORKERS = 3;
- private static final String CONNECTOR_NAME = "test-connector";
- private static final String TOPIC = "test-topic";
private static final int NUM_TASKS = 2;
private static final int NUM_RECORDS_PER_PARTITION = 10;
- private Map<String, String> workerProps;
- private EmbeddedConnectCluster.Builder connectBuilder;
+ private static final Map<Map<String, String>, EmbeddedConnectCluster> CONNECT_CLUSTERS = new ConcurrentHashMap<>();
+ @Rule
+ public TestName currentTest = new TestName();
private EmbeddedConnectCluster connect;
+ private String connectorName;
+ private String topic;
@Before
public void setup() {
- Properties brokerProps = new Properties();
- brokerProps.put("transaction.state.log.replication.factor", "1");
- brokerProps.put("transaction.state.log.min.isr", "1");
-
- // setup Connect worker properties
- workerProps = new HashMap<>();
- workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
-
- // build a Connect cluster backed by Kafka and Zk
- connectBuilder = new EmbeddedConnectCluster.Builder()
- .name("connect-cluster")
- .numWorkers(NUM_WORKERS)
- .brokerProps(brokerProps)
- .workerProps(workerProps);
+ connectorName = currentTest.getMethodName();
+ topic = currentTest.getMethodName();
+ connect = defaultConnectCluster();
}
@After
public void tearDown() {
- connect.stop();
+ Set<String> remainingConnectors = new HashSet<>(connect.connectors());
+ if (remainingConnectors.remove(connectorName)) {
+ connect.deleteConnector(connectorName);
+ }
+ try {
Review Comment:
Ah, got it. Yeah, the idea behind the assertion was to catch any stray connectors that may be left behind by future tests. It's pointless now but it may be a nice guardrail later on.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on PR #14966:
URL: https://github.com/apache/kafka/pull/14966#issuecomment-1883271039
Failures appear to be unrelated, merging...
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on code in PR #14966:
URL: https://github.com/apache/kafka/pull/14966#discussion_r1424389303
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -70,74 +80,115 @@ public class OffsetsApiIntegrationTest {
private static final long OFFSET_COMMIT_INTERVAL_MS = TimeUnit.SECONDS.toMillis(1);
private static final long OFFSET_READ_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(30);
private static final int NUM_WORKERS = 3;
- private static final String CONNECTOR_NAME = "test-connector";
- private static final String TOPIC = "test-topic";
private static final int NUM_TASKS = 2;
private static final int NUM_RECORDS_PER_PARTITION = 10;
- private Map<String, String> workerProps;
- private EmbeddedConnectCluster.Builder connectBuilder;
+ private static final Map<Map<String, String>, EmbeddedConnectCluster> CONNECT_CLUSTERS = new ConcurrentHashMap<>();
+ @Rule
+ public TestName currentTest = new TestName();
private EmbeddedConnectCluster connect;
+ private String connectorName;
+ private String topic;
@Before
public void setup() {
- Properties brokerProps = new Properties();
- brokerProps.put("transaction.state.log.replication.factor", "1");
- brokerProps.put("transaction.state.log.min.isr", "1");
-
- // setup Connect worker properties
- workerProps = new HashMap<>();
- workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
-
- // build a Connect cluster backed by Kafka and Zk
- connectBuilder = new EmbeddedConnectCluster.Builder()
- .name("connect-cluster")
- .numWorkers(NUM_WORKERS)
- .brokerProps(brokerProps)
- .workerProps(workerProps);
+ connectorName = currentTest.getMethodName();
+ topic = currentTest.getMethodName();
+ connect = defaultConnectCluster();
}
@After
public void tearDown() {
- connect.stop();
+ Set<String> remainingConnectors = new HashSet<>(connect.connectors());
+ if (remainingConnectors.remove(connectorName)) {
+ connect.deleteConnector(connectorName);
+ }
+ try {
+ assertEquals(
+ "Some connectors were not properly cleaned up after this test",
+ Collections.emptySet(),
+ remainingConnectors
+ );
+ } finally {
+ // Make a last-ditch effort to clean up the leaked connectors
+ // so as not to interfere with other test cases
+ remainingConnectors.forEach(connect::deleteConnector);
+ }
+ }
+
+ @AfterClass
+ public static void close() {
+ // stop all Connect, Kafka and Zk threads.
+ CONNECT_CLUSTERS.values().forEach(EmbeddedConnectCluster::stop);
+ }
+
+ private static EmbeddedConnectCluster createOrReuseConnectWithWorkerProps(Map<String, String> workerProps) {
+ return CONNECT_CLUSTERS.computeIfAbsent(workerProps, props -> {
+ Properties brokerProps = new Properties();
+ brokerProps.put("transaction.state.log.replication.factor", "1");
+ brokerProps.put("transaction.state.log.min.isr", "1");
+
+ // Have to declare a new map since the passed-in one may be immutable
+ Map<String, String> workerPropsWithDefaults = new HashMap<>(workerProps);
+ // Enable fast offset commits by default
+ workerPropsWithDefaults.putIfAbsent(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
+
+ EmbeddedConnectCluster result = new EmbeddedConnectCluster.Builder()
Review Comment:
Ok, I saw `connect` in `ConnectRestartAPI` being used, that's why thought we could use the same name. We can retain this.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
Posted by "sudeshwasnik (via GitHub)" <gi...@apache.org>.
sudeshwasnik commented on code in PR #14966:
URL: https://github.com/apache/kafka/pull/14966#discussion_r1421394339
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -70,74 +80,115 @@ public class OffsetsApiIntegrationTest {
private static final long OFFSET_COMMIT_INTERVAL_MS = TimeUnit.SECONDS.toMillis(1);
private static final long OFFSET_READ_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(30);
private static final int NUM_WORKERS = 3;
- private static final String CONNECTOR_NAME = "test-connector";
- private static final String TOPIC = "test-topic";
private static final int NUM_TASKS = 2;
private static final int NUM_RECORDS_PER_PARTITION = 10;
- private Map<String, String> workerProps;
- private EmbeddedConnectCluster.Builder connectBuilder;
+ private static final Map<Map<String, String>, EmbeddedConnectCluster> CONNECT_CLUSTERS = new ConcurrentHashMap<>();
+ @Rule
+ public TestName currentTest = new TestName();
private EmbeddedConnectCluster connect;
+ private String connectorName;
+ private String topic;
@Before
public void setup() {
- Properties brokerProps = new Properties();
- brokerProps.put("transaction.state.log.replication.factor", "1");
- brokerProps.put("transaction.state.log.min.isr", "1");
-
- // setup Connect worker properties
- workerProps = new HashMap<>();
- workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
-
- // build a Connect cluster backed by Kafka and Zk
- connectBuilder = new EmbeddedConnectCluster.Builder()
- .name("connect-cluster")
- .numWorkers(NUM_WORKERS)
- .brokerProps(brokerProps)
- .workerProps(workerProps);
+ connectorName = currentTest.getMethodName();
Review Comment:
For tests that have common substring in the beginning (testAlterSourceConnectorOffsetsExactlyOnceSupportEnabled and testAlterSourceConnectorOffsets), can a check done with `startsWith` [here](https://github.com/apache/kafka/blob/trunk/connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java#L954) result in an incorrect verification ?
ie, offsets written by testAlterSourceConnectorOffsetsExactlyOnceSupportEnabled test will `startWith` `testAlterSourceConnectorOffsets`, so we may incorrectly verify `testAlterSourceConnectorOffsets` to be passing.
This may never happen in reality and is a corner case, but wdyt about clearing all offsets topic after each test? Or we can replace `startsWith` with string-equality check too..
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -70,74 +80,115 @@ public class OffsetsApiIntegrationTest {
private static final long OFFSET_COMMIT_INTERVAL_MS = TimeUnit.SECONDS.toMillis(1);
private static final long OFFSET_READ_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(30);
private static final int NUM_WORKERS = 3;
- private static final String CONNECTOR_NAME = "test-connector";
- private static final String TOPIC = "test-topic";
private static final int NUM_TASKS = 2;
private static final int NUM_RECORDS_PER_PARTITION = 10;
- private Map<String, String> workerProps;
- private EmbeddedConnectCluster.Builder connectBuilder;
+ private static final Map<Map<String, String>, EmbeddedConnectCluster> CONNECT_CLUSTERS = new ConcurrentHashMap<>();
+ @Rule
+ public TestName currentTest = new TestName();
private EmbeddedConnectCluster connect;
+ private String connectorName;
+ private String topic;
@Before
public void setup() {
- Properties brokerProps = new Properties();
- brokerProps.put("transaction.state.log.replication.factor", "1");
- brokerProps.put("transaction.state.log.min.isr", "1");
-
- // setup Connect worker properties
- workerProps = new HashMap<>();
- workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
-
- // build a Connect cluster backed by Kafka and Zk
- connectBuilder = new EmbeddedConnectCluster.Builder()
- .name("connect-cluster")
- .numWorkers(NUM_WORKERS)
- .brokerProps(brokerProps)
- .workerProps(workerProps);
+ connectorName = currentTest.getMethodName();
+ topic = currentTest.getMethodName();
+ connect = defaultConnectCluster();
}
@After
public void tearDown() {
- connect.stop();
+ Set<String> remainingConnectors = new HashSet<>(connect.connectors());
+ if (remainingConnectors.remove(connectorName)) {
+ connect.deleteConnector(connectorName);
+ }
+ try {
Review Comment:
```suggestion
}
remainingConnectors = new HashSet<>(connect.connectors());
try {
```
wdyt ?
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #14966:
URL: https://github.com/apache/kafka/pull/14966#discussion_r1425557390
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -895,6 +909,55 @@ private Map<String, String> baseSourceConnectorConfigs() {
return props;
}
+ /**
+ * Modify (i.e., alter or reset) the offsets for a sink connector, with retry logic to
+ * handle cases where laggy task shutdown may have left a consumer in the group.
+ * @param offsetsToAlter the offsets to alter for the sink connector, or null if
+ * the connector's offets should be reset instead
+ * @return the response from the REST API, if the request was successful
+ * @throws InterruptedException if the thread is interrupted while waiting for a
+ * request to modify the connector's offsets to succeed
+ * @see <a href="https://issues.apache.org/jira/browse/KAFKA-15826">KAFKA-15826</a>
+ */
+ private String modifySinkConnectorOffsetsWithRetry(ConnectorOffsets offsetsToAlter) throws InterruptedException {
+ // Some retry logic is necessary to account for KAFKA-15826,
+ // where laggy sink task startup/shutdown can leave consumers running
+ String conditionDetails = "Failed to "
+ + (offsetsToAlter != null ? "alter" : "reset")
+ + " sink connector offsets in time";
+ AtomicReference<String> responseReference = new AtomicReference<>();
+ waitForCondition(
+ () -> {
+ try {
+ if (offsetsToAlter == null) {
+ responseReference.set(connect.resetConnectorOffsets(connectorName));
+ } else {
+ responseReference.set(connect.alterConnectorOffsets(connectorName, offsetsToAlter));
+ }
+ return true;
+ } catch (ConnectRestException e) {
+ boolean internalServerError = e.statusCode() == INTERNAL_SERVER_ERROR.getStatusCode();
+
+ String message = Optional.of(e.getMessage()).orElse("");
+ boolean failedToResetConsumerOffsets = message.contains("Failed to reset consumer group offsets for connector");
Review Comment:
Ah yep, good catch! Thanks.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante merged PR #14966:
URL: https://github.com/apache/kafka/pull/14966
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on PR #14966:
URL: https://github.com/apache/kafka/pull/14966#issuecomment-1852343242
Thanks Chris, this looks like a really nice improvement! I can review sometime later this week.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
Posted by "sudeshwasnik (via GitHub)" <gi...@apache.org>.
sudeshwasnik commented on code in PR #14966:
URL: https://github.com/apache/kafka/pull/14966#discussion_r1424444423
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -70,74 +80,115 @@ public class OffsetsApiIntegrationTest {
private static final long OFFSET_COMMIT_INTERVAL_MS = TimeUnit.SECONDS.toMillis(1);
private static final long OFFSET_READ_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(30);
private static final int NUM_WORKERS = 3;
- private static final String CONNECTOR_NAME = "test-connector";
- private static final String TOPIC = "test-topic";
private static final int NUM_TASKS = 2;
private static final int NUM_RECORDS_PER_PARTITION = 10;
- private Map<String, String> workerProps;
- private EmbeddedConnectCluster.Builder connectBuilder;
+ private static final Map<Map<String, String>, EmbeddedConnectCluster> CONNECT_CLUSTERS = new ConcurrentHashMap<>();
+ @Rule
+ public TestName currentTest = new TestName();
private EmbeddedConnectCluster connect;
+ private String connectorName;
+ private String topic;
@Before
public void setup() {
- Properties brokerProps = new Properties();
- brokerProps.put("transaction.state.log.replication.factor", "1");
- brokerProps.put("transaction.state.log.min.isr", "1");
-
- // setup Connect worker properties
- workerProps = new HashMap<>();
- workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
-
- // build a Connect cluster backed by Kafka and Zk
- connectBuilder = new EmbeddedConnectCluster.Builder()
- .name("connect-cluster")
- .numWorkers(NUM_WORKERS)
- .brokerProps(brokerProps)
- .workerProps(workerProps);
+ connectorName = currentTest.getMethodName();
+ topic = currentTest.getMethodName();
+ connect = defaultConnectCluster();
}
@After
public void tearDown() {
- connect.stop();
+ Set<String> remainingConnectors = new HashSet<>(connect.connectors());
+ if (remainingConnectors.remove(connectorName)) {
+ connect.deleteConnector(connectorName);
+ }
+ try {
Review Comment:
line 102 -> `remainingConnectors.remove(connectorName)` will remove elements anyways, so line-106's assertEquals will always be true - so the assertion is pointless. Which I thought wasn't okayy.
but your point is valid too! feel free to resolve this comment
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #14966:
URL: https://github.com/apache/kafka/pull/14966#discussion_r1423199416
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -324,127 +363,151 @@ public void testAlterSinkConnectorOffsetsDifferentKafkaClusterTargeted() throws
private void alterAndVerifySinkConnectorOffsets(Map<String, String> connectorConfigs, EmbeddedKafkaCluster kafkaCluster) throws Exception {
int numPartitions = 3;
- kafkaCluster.createTopic(TOPIC, numPartitions);
+ kafkaCluster.createTopic(topic, numPartitions);
// Produce records to each partition
for (int partition = 0; partition < numPartitions; partition++) {
for (int record = 0; record < NUM_RECORDS_PER_PARTITION; record++) {
- kafkaCluster.produce(TOPIC, partition, "key", "value");
+ kafkaCluster.produce(topic, partition, "key", "value");
}
}
// Create sink connector
- connect.configureConnector(CONNECTOR_NAME, connectorConfigs);
- connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(CONNECTOR_NAME, NUM_TASKS,
+ connect.configureConnector(connectorName, connectorConfigs);
+ connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(connectorName, NUM_TASKS,
"Connector tasks did not start in time.");
- verifyExpectedSinkConnectorOffsets(CONNECTOR_NAME, TOPIC, numPartitions, NUM_RECORDS_PER_PARTITION,
+ verifyExpectedSinkConnectorOffsets(connectorName, topic, numPartitions, NUM_RECORDS_PER_PARTITION,
"Sink connector consumer group offsets should catch up to the topic end offsets");
- connect.stopConnector(CONNECTOR_NAME);
+ connect.stopConnector(connectorName);
connect.assertions().assertConnectorIsStopped(
- CONNECTOR_NAME,
+ connectorName,
"Connector did not stop in time"
);
// Delete the offset of one partition; alter the offsets of the others
List<ConnectorOffset> offsetsToAlter = new ArrayList<>();
Map<String, Object> partition = new HashMap<>();
- partition.put(SinkUtils.KAFKA_TOPIC_KEY, TOPIC);
+ partition.put(SinkUtils.KAFKA_TOPIC_KEY, topic);
partition.put(SinkUtils.KAFKA_PARTITION_KEY, 0);
offsetsToAlter.add(new ConnectorOffset(partition, null));
for (int i = 1; i < numPartitions; i++) {
partition = new HashMap<>();
- partition.put(SinkUtils.KAFKA_TOPIC_KEY, TOPIC);
+ partition.put(SinkUtils.KAFKA_TOPIC_KEY, topic);
partition.put(SinkUtils.KAFKA_PARTITION_KEY, i);
offsetsToAlter.add(new ConnectorOffset(partition, Collections.singletonMap(SinkUtils.KAFKA_OFFSET_KEY, 5)));
}
- String response = connect.alterConnectorOffsets(CONNECTOR_NAME, new ConnectorOffsets(offsetsToAlter));
+ AtomicReference<String> responseReference = new AtomicReference<>();
+ // Some retry logic is necessary to account for KAFKA-15826,
+ // where laggy sink task startup/shutdown can leave consumers running
+ waitForCondition(
+ () -> {
+ try {
+ responseReference.set(connect.alterConnectorOffsets(connectorName, new ConnectorOffsets(offsetsToAlter)));
+ return true;
+ } catch (ConnectRestException e) {
+ boolean internalServerError = e.statusCode() == INTERNAL_SERVER_ERROR.getStatusCode();
+
+ String message = Optional.of(e.getMessage()).orElse("");
+ boolean failedToResetConsumerOffsets = message.contains("Failed to reset consumer group offsets for connector");
+ boolean canBeRetried = message.contains("If the connector is in a stopped state, this operation can be safely retried");
+
+ boolean retriable = internalServerError && failedToResetConsumerOffsets && canBeRetried;
+ if (retriable) {
+ return false;
+ } else {
+ throw new NoRetryException(e);
+ }
Review Comment:
Fewer lines doesn't necessarily mean simpler; I think this is fine as-is.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -70,74 +80,115 @@ public class OffsetsApiIntegrationTest {
private static final long OFFSET_COMMIT_INTERVAL_MS = TimeUnit.SECONDS.toMillis(1);
private static final long OFFSET_READ_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(30);
private static final int NUM_WORKERS = 3;
- private static final String CONNECTOR_NAME = "test-connector";
- private static final String TOPIC = "test-topic";
private static final int NUM_TASKS = 2;
private static final int NUM_RECORDS_PER_PARTITION = 10;
- private Map<String, String> workerProps;
- private EmbeddedConnectCluster.Builder connectBuilder;
+ private static final Map<Map<String, String>, EmbeddedConnectCluster> CONNECT_CLUSTERS = new ConcurrentHashMap<>();
+ @Rule
+ public TestName currentTest = new TestName();
private EmbeddedConnectCluster connect;
+ private String connectorName;
+ private String topic;
@Before
public void setup() {
- Properties brokerProps = new Properties();
- brokerProps.put("transaction.state.log.replication.factor", "1");
- brokerProps.put("transaction.state.log.min.isr", "1");
-
- // setup Connect worker properties
- workerProps = new HashMap<>();
- workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
-
- // build a Connect cluster backed by Kafka and Zk
- connectBuilder = new EmbeddedConnectCluster.Builder()
- .name("connect-cluster")
- .numWorkers(NUM_WORKERS)
- .brokerProps(brokerProps)
- .workerProps(workerProps);
+ connectorName = currentTest.getMethodName();
+ topic = currentTest.getMethodName();
+ connect = defaultConnectCluster();
}
@After
public void tearDown() {
- connect.stop();
+ Set<String> remainingConnectors = new HashSet<>(connect.connectors());
+ if (remainingConnectors.remove(connectorName)) {
+ connect.deleteConnector(connectorName);
+ }
+ try {
+ assertEquals(
+ "Some connectors were not properly cleaned up after this test",
+ Collections.emptySet(),
+ remainingConnectors
+ );
+ } finally {
+ // Make a last-ditch effort to clean up the leaked connectors
+ // so as not to interfere with other test cases
+ remainingConnectors.forEach(connect::deleteConnector);
+ }
+ }
+
+ @AfterClass
+ public static void close() {
+ // stop all Connect, Kafka and Zk threads.
+ CONNECT_CLUSTERS.values().forEach(EmbeddedConnectCluster::stop);
+ }
+
+ private static EmbeddedConnectCluster createOrReuseConnectWithWorkerProps(Map<String, String> workerProps) {
+ return CONNECT_CLUSTERS.computeIfAbsent(workerProps, props -> {
+ Properties brokerProps = new Properties();
+ brokerProps.put("transaction.state.log.replication.factor", "1");
+ brokerProps.put("transaction.state.log.min.isr", "1");
+
+ // Have to declare a new map since the passed-in one may be immutable
+ Map<String, String> workerPropsWithDefaults = new HashMap<>(workerProps);
+ // Enable fast offset commits by default
+ workerPropsWithDefaults.putIfAbsent(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
+
+ EmbeddedConnectCluster result = new EmbeddedConnectCluster.Builder()
+ .name("connect-cluster")
+ .numWorkers(NUM_WORKERS)
+ .brokerProps(brokerProps)
+ .workerProps(workerPropsWithDefaults)
+ .build();
+
+ result.start();
+
+ return result;
+ });
+ }
+
+ private static EmbeddedConnectCluster defaultConnectCluster() {
+ return createOrReuseConnectWithWorkerProps(Collections.emptyMap());
+ }
+
+ private static EmbeddedConnectCluster exactlyOnceSourceConnectCluster() {
+ Map<String, String> workerProps = Collections.singletonMap(
+ DistributedConfig.EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG,
+ "enabled"
+ );
+ return createOrReuseConnectWithWorkerProps(workerProps);
}
@Test
public void testGetNonExistentConnectorOffsets() {
- connect = connectBuilder.build();
- connect.start();
ConnectRestException e = assertThrows(ConnectRestException.class,
() -> connect.connectorOffsets("non-existent-connector"));
assertEquals(404, e.errorCode());
}
@Test
public void testGetSinkConnectorOffsets() throws Exception {
- connect = connectBuilder.build();
- connect.start();
getAndVerifySinkConnectorOffsets(baseSinkConnectorConfigs(), connect.kafka());
}
@Test
public void testGetSinkConnectorOffsetsOverriddenConsumerGroupId() throws Exception {
- connect = connectBuilder.build();
- connect.start();
Map<String, String> connectorConfigs = baseSinkConnectorConfigs();
- connectorConfigs.put(ConnectorConfig.CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX + CommonClientConfigs.GROUP_ID_CONFIG,
- "overridden-group-id");
+ String overriddenGroupId = connectorName = "-overridden-group-id";
Review Comment:
There's no need to use the same string in different test cases, so there's very little benefit to extracting this out to a class constant. We could just as easily change it to `"custom-group-id"`, `"other-group-id"`, etc. in a single test and things would work just as well.
It also adds noise to the entire test suite to add class-level constants; since overridden group IDs are only used in three of the twenty-seven tests here, IMO it's cleaner to keep these strings isolated to the tests that actually need them.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #14966:
URL: https://github.com/apache/kafka/pull/14966#discussion_r1423199462
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -70,74 +80,115 @@ public class OffsetsApiIntegrationTest {
private static final long OFFSET_COMMIT_INTERVAL_MS = TimeUnit.SECONDS.toMillis(1);
private static final long OFFSET_READ_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(30);
private static final int NUM_WORKERS = 3;
- private static final String CONNECTOR_NAME = "test-connector";
- private static final String TOPIC = "test-topic";
private static final int NUM_TASKS = 2;
private static final int NUM_RECORDS_PER_PARTITION = 10;
- private Map<String, String> workerProps;
- private EmbeddedConnectCluster.Builder connectBuilder;
+ private static final Map<Map<String, String>, EmbeddedConnectCluster> CONNECT_CLUSTERS = new ConcurrentHashMap<>();
+ @Rule
+ public TestName currentTest = new TestName();
private EmbeddedConnectCluster connect;
+ private String connectorName;
+ private String topic;
@Before
public void setup() {
- Properties brokerProps = new Properties();
- brokerProps.put("transaction.state.log.replication.factor", "1");
- brokerProps.put("transaction.state.log.min.isr", "1");
-
- // setup Connect worker properties
- workerProps = new HashMap<>();
- workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
-
- // build a Connect cluster backed by Kafka and Zk
- connectBuilder = new EmbeddedConnectCluster.Builder()
- .name("connect-cluster")
- .numWorkers(NUM_WORKERS)
- .brokerProps(brokerProps)
- .workerProps(workerProps);
+ connectorName = currentTest.getMethodName();
+ topic = currentTest.getMethodName();
+ connect = defaultConnectCluster();
}
@After
public void tearDown() {
- connect.stop();
+ Set<String> remainingConnectors = new HashSet<>(connect.connectors());
+ if (remainingConnectors.remove(connectorName)) {
+ connect.deleteConnector(connectorName);
+ }
+ try {
+ assertEquals(
+ "Some connectors were not properly cleaned up after this test",
+ Collections.emptySet(),
+ remainingConnectors
+ );
+ } finally {
+ // Make a last-ditch effort to clean up the leaked connectors
+ // so as not to interfere with other test cases
+ remainingConnectors.forEach(connect::deleteConnector);
+ }
+ }
+
+ @AfterClass
+ public static void close() {
+ // stop all Connect, Kafka and Zk threads.
+ CONNECT_CLUSTERS.values().forEach(EmbeddedConnectCluster::stop);
+ }
+
+ private static EmbeddedConnectCluster createOrReuseConnectWithWorkerProps(Map<String, String> workerProps) {
+ return CONNECT_CLUSTERS.computeIfAbsent(workerProps, props -> {
+ Properties brokerProps = new Properties();
+ brokerProps.put("transaction.state.log.replication.factor", "1");
+ brokerProps.put("transaction.state.log.min.isr", "1");
+
+ // Have to declare a new map since the passed-in one may be immutable
+ Map<String, String> workerPropsWithDefaults = new HashMap<>(workerProps);
+ // Enable fast offset commits by default
+ workerPropsWithDefaults.putIfAbsent(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
+
+ EmbeddedConnectCluster result = new EmbeddedConnectCluster.Builder()
+ .name("connect-cluster")
+ .numWorkers(NUM_WORKERS)
+ .brokerProps(brokerProps)
+ .workerProps(workerPropsWithDefaults)
+ .build();
+
+ result.start();
+
+ return result;
+ });
+ }
+
+ private static EmbeddedConnectCluster defaultConnectCluster() {
+ return createOrReuseConnectWithWorkerProps(Collections.emptyMap());
+ }
+
+ private static EmbeddedConnectCluster exactlyOnceSourceConnectCluster() {
+ Map<String, String> workerProps = Collections.singletonMap(
+ DistributedConfig.EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG,
+ "enabled"
+ );
+ return createOrReuseConnectWithWorkerProps(workerProps);
}
@Test
public void testGetNonExistentConnectorOffsets() {
- connect = connectBuilder.build();
- connect.start();
ConnectRestException e = assertThrows(ConnectRestException.class,
() -> connect.connectorOffsets("non-existent-connector"));
assertEquals(404, e.errorCode());
}
@Test
public void testGetSinkConnectorOffsets() throws Exception {
- connect = connectBuilder.build();
- connect.start();
getAndVerifySinkConnectorOffsets(baseSinkConnectorConfigs(), connect.kafka());
}
@Test
public void testGetSinkConnectorOffsetsOverriddenConsumerGroupId() throws Exception {
- connect = connectBuilder.build();
- connect.start();
Map<String, String> connectorConfigs = baseSinkConnectorConfigs();
- connectorConfigs.put(ConnectorConfig.CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX + CommonClientConfigs.GROUP_ID_CONFIG,
- "overridden-group-id");
+ String overriddenGroupId = connectorName = "-overridden-group-id";
Review Comment:
There's no need to use the same string in different test cases, so there's less benefit than usual for extracting this out to a class constant. We could just as easily change it to `"custom-group-id"`, `"other-group-id"`, etc. in a single test and things would work just as well.
It also adds noise to the entire test suite to add class-level constants; since overridden group IDs are only used in three of the twenty-seven tests here, IMO it's cleaner to keep these strings isolated to the tests that actually need them.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -70,74 +80,115 @@ public class OffsetsApiIntegrationTest {
private static final long OFFSET_COMMIT_INTERVAL_MS = TimeUnit.SECONDS.toMillis(1);
private static final long OFFSET_READ_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(30);
private static final int NUM_WORKERS = 3;
- private static final String CONNECTOR_NAME = "test-connector";
- private static final String TOPIC = "test-topic";
private static final int NUM_TASKS = 2;
private static final int NUM_RECORDS_PER_PARTITION = 10;
- private Map<String, String> workerProps;
- private EmbeddedConnectCluster.Builder connectBuilder;
+ private static final Map<Map<String, String>, EmbeddedConnectCluster> CONNECT_CLUSTERS = new ConcurrentHashMap<>();
+ @Rule
+ public TestName currentTest = new TestName();
private EmbeddedConnectCluster connect;
+ private String connectorName;
+ private String topic;
@Before
public void setup() {
- Properties brokerProps = new Properties();
- brokerProps.put("transaction.state.log.replication.factor", "1");
- brokerProps.put("transaction.state.log.min.isr", "1");
-
- // setup Connect worker properties
- workerProps = new HashMap<>();
- workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
-
- // build a Connect cluster backed by Kafka and Zk
- connectBuilder = new EmbeddedConnectCluster.Builder()
- .name("connect-cluster")
- .numWorkers(NUM_WORKERS)
- .brokerProps(brokerProps)
- .workerProps(workerProps);
+ connectorName = currentTest.getMethodName();
+ topic = currentTest.getMethodName();
+ connect = defaultConnectCluster();
}
@After
public void tearDown() {
- connect.stop();
+ Set<String> remainingConnectors = new HashSet<>(connect.connectors());
+ if (remainingConnectors.remove(connectorName)) {
+ connect.deleteConnector(connectorName);
+ }
+ try {
+ assertEquals(
+ "Some connectors were not properly cleaned up after this test",
+ Collections.emptySet(),
+ remainingConnectors
+ );
+ } finally {
+ // Make a last-ditch effort to clean up the leaked connectors
+ // so as not to interfere with other test cases
+ remainingConnectors.forEach(connect::deleteConnector);
+ }
+ }
+
+ @AfterClass
+ public static void close() {
+ // stop all Connect, Kafka and Zk threads.
+ CONNECT_CLUSTERS.values().forEach(EmbeddedConnectCluster::stop);
+ }
+
+ private static EmbeddedConnectCluster createOrReuseConnectWithWorkerProps(Map<String, String> workerProps) {
+ return CONNECT_CLUSTERS.computeIfAbsent(workerProps, props -> {
+ Properties brokerProps = new Properties();
+ brokerProps.put("transaction.state.log.replication.factor", "1");
+ brokerProps.put("transaction.state.log.min.isr", "1");
+
+ // Have to declare a new map since the passed-in one may be immutable
+ Map<String, String> workerPropsWithDefaults = new HashMap<>(workerProps);
+ // Enable fast offset commits by default
+ workerPropsWithDefaults.putIfAbsent(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
+
+ EmbeddedConnectCluster result = new EmbeddedConnectCluster.Builder()
+ .name("connect-cluster")
+ .numWorkers(NUM_WORKERS)
+ .brokerProps(brokerProps)
+ .workerProps(workerPropsWithDefaults)
+ .build();
+
+ result.start();
+
+ return result;
+ });
+ }
+
+ private static EmbeddedConnectCluster defaultConnectCluster() {
+ return createOrReuseConnectWithWorkerProps(Collections.emptyMap());
+ }
+
+ private static EmbeddedConnectCluster exactlyOnceSourceConnectCluster() {
+ Map<String, String> workerProps = Collections.singletonMap(
+ DistributedConfig.EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG,
+ "enabled"
+ );
+ return createOrReuseConnectWithWorkerProps(workerProps);
}
@Test
public void testGetNonExistentConnectorOffsets() {
- connect = connectBuilder.build();
- connect.start();
ConnectRestException e = assertThrows(ConnectRestException.class,
() -> connect.connectorOffsets("non-existent-connector"));
assertEquals(404, e.errorCode());
}
@Test
public void testGetSinkConnectorOffsets() throws Exception {
- connect = connectBuilder.build();
- connect.start();
getAndVerifySinkConnectorOffsets(baseSinkConnectorConfigs(), connect.kafka());
}
@Test
public void testGetSinkConnectorOffsetsOverriddenConsumerGroupId() throws Exception {
- connect = connectBuilder.build();
- connect.start();
Map<String, String> connectorConfigs = baseSinkConnectorConfigs();
- connectorConfigs.put(ConnectorConfig.CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX + CommonClientConfigs.GROUP_ID_CONFIG,
- "overridden-group-id");
+ String overriddenGroupId = connectorName = "-overridden-group-id";
Review Comment:
There's no need to use the same string in different test cases, so there's less benefit than usual for extracting this out to a class constant. We could change it to `"custom-group-id"`, `"other-group-id"`, etc. in a single test and things would work just as well.
It also adds noise to the entire test suite to add class-level constants; since overridden group IDs are only used in three of the twenty-seven tests here, IMO it's cleaner to keep these strings isolated to the tests that actually need them.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
Posted by "sudeshwasnik (via GitHub)" <gi...@apache.org>.
sudeshwasnik commented on code in PR #14966:
URL: https://github.com/apache/kafka/pull/14966#discussion_r1424444423
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -70,74 +80,115 @@ public class OffsetsApiIntegrationTest {
private static final long OFFSET_COMMIT_INTERVAL_MS = TimeUnit.SECONDS.toMillis(1);
private static final long OFFSET_READ_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(30);
private static final int NUM_WORKERS = 3;
- private static final String CONNECTOR_NAME = "test-connector";
- private static final String TOPIC = "test-topic";
private static final int NUM_TASKS = 2;
private static final int NUM_RECORDS_PER_PARTITION = 10;
- private Map<String, String> workerProps;
- private EmbeddedConnectCluster.Builder connectBuilder;
+ private static final Map<Map<String, String>, EmbeddedConnectCluster> CONNECT_CLUSTERS = new ConcurrentHashMap<>();
+ @Rule
+ public TestName currentTest = new TestName();
private EmbeddedConnectCluster connect;
+ private String connectorName;
+ private String topic;
@Before
public void setup() {
- Properties brokerProps = new Properties();
- brokerProps.put("transaction.state.log.replication.factor", "1");
- brokerProps.put("transaction.state.log.min.isr", "1");
-
- // setup Connect worker properties
- workerProps = new HashMap<>();
- workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
-
- // build a Connect cluster backed by Kafka and Zk
- connectBuilder = new EmbeddedConnectCluster.Builder()
- .name("connect-cluster")
- .numWorkers(NUM_WORKERS)
- .brokerProps(brokerProps)
- .workerProps(workerProps);
+ connectorName = currentTest.getMethodName();
+ topic = currentTest.getMethodName();
+ connect = defaultConnectCluster();
}
@After
public void tearDown() {
- connect.stop();
+ Set<String> remainingConnectors = new HashSet<>(connect.connectors());
+ if (remainingConnectors.remove(connectorName)) {
+ connect.deleteConnector(connectorName);
+ }
+ try {
Review Comment:
line 102 -> `remainingConnectors.remove(connectorName)` will remove elements anyways, so line-106's assertEquals will always be true - so the assertion is pointless. Which I thought wasn't okayy.
but your point is valid too! feel free to resolve this comment
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #14966:
URL: https://github.com/apache/kafka/pull/14966#discussion_r1425647269
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -895,6 +909,55 @@ private Map<String, String> baseSourceConnectorConfigs() {
return props;
}
+ /**
+ * Modify (i.e., alter or reset) the offsets for a sink connector, with retry logic to
+ * handle cases where laggy task shutdown may have left a consumer in the group.
+ * @param offsetsToAlter the offsets to alter for the sink connector, or null if
+ * the connector's offets should be reset instead
+ * @return the response from the REST API, if the request was successful
+ * @throws InterruptedException if the thread is interrupted while waiting for a
+ * request to modify the connector's offsets to succeed
+ * @see <a href="https://issues.apache.org/jira/browse/KAFKA-15826">KAFKA-15826</a>
+ */
+ private String modifySinkConnectorOffsetsWithRetry(ConnectorOffsets offsetsToAlter) throws InterruptedException {
+ // Some retry logic is necessary to account for KAFKA-15826,
+ // where laggy sink task startup/shutdown can leave consumers running
Review Comment:
The hypothesis I have is that our CI environment is so slow, that even `MonitorableSinkTask` instances may fail to shut down gracefully. And, because of KAFKA-15826, those task instances will still have running consumers that can interfere with requests to reset/alter consumer group offsets.
If my guess is correct, the retry logic should help because it gives the task a little more time to complete shut down and close its consumer, and it should no longer be necessary once KAFKA-15826 is resolved.
It's also possible that slow rebalances or metadata updates may cause transient errors even when all tasks have completed gracefully, though I have to admit I haven't done enough research on that part of the code base to speak with any certainty.
Does that clear things up?
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
Posted by "sudeshwasnik (via GitHub)" <gi...@apache.org>.
sudeshwasnik commented on code in PR #14966:
URL: https://github.com/apache/kafka/pull/14966#discussion_r1424446893
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -70,74 +80,115 @@ public class OffsetsApiIntegrationTest {
private static final long OFFSET_COMMIT_INTERVAL_MS = TimeUnit.SECONDS.toMillis(1);
private static final long OFFSET_READ_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(30);
private static final int NUM_WORKERS = 3;
- private static final String CONNECTOR_NAME = "test-connector";
- private static final String TOPIC = "test-topic";
private static final int NUM_TASKS = 2;
private static final int NUM_RECORDS_PER_PARTITION = 10;
- private Map<String, String> workerProps;
- private EmbeddedConnectCluster.Builder connectBuilder;
+ private static final Map<Map<String, String>, EmbeddedConnectCluster> CONNECT_CLUSTERS = new ConcurrentHashMap<>();
+ @Rule
+ public TestName currentTest = new TestName();
private EmbeddedConnectCluster connect;
+ private String connectorName;
+ private String topic;
@Before
public void setup() {
- Properties brokerProps = new Properties();
- brokerProps.put("transaction.state.log.replication.factor", "1");
- brokerProps.put("transaction.state.log.min.isr", "1");
-
- // setup Connect worker properties
- workerProps = new HashMap<>();
- workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
-
- // build a Connect cluster backed by Kafka and Zk
- connectBuilder = new EmbeddedConnectCluster.Builder()
- .name("connect-cluster")
- .numWorkers(NUM_WORKERS)
- .brokerProps(brokerProps)
- .workerProps(workerProps);
+ connectorName = currentTest.getMethodName();
+ topic = currentTest.getMethodName();
+ connect = defaultConnectCluster();
}
@After
public void tearDown() {
- connect.stop();
+ Set<String> remainingConnectors = new HashSet<>(connect.connectors());
+ if (remainingConnectors.remove(connectorName)) {
+ connect.deleteConnector(connectorName);
+ }
+ try {
Review Comment:
sounds good ! thanks !
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #14966:
URL: https://github.com/apache/kafka/pull/14966#discussion_r1424910127
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -70,74 +80,115 @@ public class OffsetsApiIntegrationTest {
private static final long OFFSET_COMMIT_INTERVAL_MS = TimeUnit.SECONDS.toMillis(1);
private static final long OFFSET_READ_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(30);
private static final int NUM_WORKERS = 3;
- private static final String CONNECTOR_NAME = "test-connector";
- private static final String TOPIC = "test-topic";
private static final int NUM_TASKS = 2;
private static final int NUM_RECORDS_PER_PARTITION = 10;
- private Map<String, String> workerProps;
- private EmbeddedConnectCluster.Builder connectBuilder;
+ private static final Map<Map<String, String>, EmbeddedConnectCluster> CONNECT_CLUSTERS = new ConcurrentHashMap<>();
+ @Rule
+ public TestName currentTest = new TestName();
private EmbeddedConnectCluster connect;
+ private String connectorName;
+ private String topic;
@Before
public void setup() {
- Properties brokerProps = new Properties();
- brokerProps.put("transaction.state.log.replication.factor", "1");
- brokerProps.put("transaction.state.log.min.isr", "1");
-
- // setup Connect worker properties
- workerProps = new HashMap<>();
- workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
-
- // build a Connect cluster backed by Kafka and Zk
- connectBuilder = new EmbeddedConnectCluster.Builder()
- .name("connect-cluster")
- .numWorkers(NUM_WORKERS)
- .brokerProps(brokerProps)
- .workerProps(workerProps);
+ connectorName = currentTest.getMethodName();
+ topic = currentTest.getMethodName();
+ connect = defaultConnectCluster();
}
@After
public void tearDown() {
- connect.stop();
+ Set<String> remainingConnectors = new HashSet<>(connect.connectors());
+ if (remainingConnectors.remove(connectorName)) {
+ connect.deleteConnector(connectorName);
+ }
+ try {
+ assertEquals(
+ "Some connectors were not properly cleaned up after this test",
+ Collections.emptySet(),
+ remainingConnectors
+ );
+ } finally {
+ // Make a last-ditch effort to clean up the leaked connectors
+ // so as not to interfere with other test cases
+ remainingConnectors.forEach(connect::deleteConnector);
+ }
+ }
+
+ @AfterClass
+ public static void close() {
+ // stop all Connect, Kafka and Zk threads.
+ CONNECT_CLUSTERS.values().forEach(EmbeddedConnectCluster::stop);
+ }
+
+ private static EmbeddedConnectCluster createOrReuseConnectWithWorkerProps(Map<String, String> workerProps) {
+ return CONNECT_CLUSTERS.computeIfAbsent(workerProps, props -> {
+ Properties brokerProps = new Properties();
+ brokerProps.put("transaction.state.log.replication.factor", "1");
+ brokerProps.put("transaction.state.log.min.isr", "1");
+
+ // Have to declare a new map since the passed-in one may be immutable
+ Map<String, String> workerPropsWithDefaults = new HashMap<>(workerProps);
+ // Enable fast offset commits by default
+ workerPropsWithDefaults.putIfAbsent(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
+
+ EmbeddedConnectCluster result = new EmbeddedConnectCluster.Builder()
+ .name("connect-cluster")
+ .numWorkers(NUM_WORKERS)
+ .brokerProps(brokerProps)
+ .workerProps(workerPropsWithDefaults)
+ .build();
+
+ result.start();
+
+ return result;
+ });
+ }
+
+ private static EmbeddedConnectCluster defaultConnectCluster() {
+ return createOrReuseConnectWithWorkerProps(Collections.emptyMap());
+ }
+
+ private static EmbeddedConnectCluster exactlyOnceSourceConnectCluster() {
+ Map<String, String> workerProps = Collections.singletonMap(
+ DistributedConfig.EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG,
+ "enabled"
+ );
+ return createOrReuseConnectWithWorkerProps(workerProps);
}
@Test
public void testGetNonExistentConnectorOffsets() {
- connect = connectBuilder.build();
- connect.start();
ConnectRestException e = assertThrows(ConnectRestException.class,
() -> connect.connectorOffsets("non-existent-connector"));
assertEquals(404, e.errorCode());
}
@Test
public void testGetSinkConnectorOffsets() throws Exception {
- connect = connectBuilder.build();
- connect.start();
getAndVerifySinkConnectorOffsets(baseSinkConnectorConfigs(), connect.kafka());
}
@Test
public void testGetSinkConnectorOffsetsOverriddenConsumerGroupId() throws Exception {
- connect = connectBuilder.build();
- connect.start();
Map<String, String> connectorConfigs = baseSinkConnectorConfigs();
- connectorConfigs.put(ConnectorConfig.CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX + CommonClientConfigs.GROUP_ID_CONFIG,
- "overridden-group-id");
+ String overriddenGroupId = connectorName = "-overridden-group-id";
Review Comment:
```suggestion
String overriddenGroupId = connectorName + "-overridden-group-id";
```
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -895,6 +909,55 @@ private Map<String, String> baseSourceConnectorConfigs() {
return props;
}
+ /**
+ * Modify (i.e., alter or reset) the offsets for a sink connector, with retry logic to
+ * handle cases where laggy task shutdown may have left a consumer in the group.
+ * @param offsetsToAlter the offsets to alter for the sink connector, or null if
+ * the connector's offets should be reset instead
+ * @return the response from the REST API, if the request was successful
+ * @throws InterruptedException if the thread is interrupted while waiting for a
+ * request to modify the connector's offsets to succeed
+ * @see <a href="https://issues.apache.org/jira/browse/KAFKA-15826">KAFKA-15826</a>
+ */
+ private String modifySinkConnectorOffsetsWithRetry(ConnectorOffsets offsetsToAlter) throws InterruptedException {
+ // Some retry logic is necessary to account for KAFKA-15826,
+ // where laggy sink task startup/shutdown can leave consumers running
Review Comment:
I'm +1 for the retry logic here since I've seen the same occasional CI flakiness that you've indicated, but I don't quite get how it's related to `KAFKA-15826` given that the `MonitorableSinkConnector` / `MonitorableSinkTask` being used here isn't really expected to block long enough in its stop method to cause task cancellation to occur? Furthermore, given that the referenced ticket isn't resolved yet, retrying here wouldn't make sense since even task cancellation wouldn't cause the task's consumer to be closed right? Or is it linked here as more of a future reference (once the ticket is resolved)?
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -895,6 +909,55 @@ private Map<String, String> baseSourceConnectorConfigs() {
return props;
}
+ /**
+ * Modify (i.e., alter or reset) the offsets for a sink connector, with retry logic to
+ * handle cases where laggy task shutdown may have left a consumer in the group.
+ * @param offsetsToAlter the offsets to alter for the sink connector, or null if
+ * the connector's offets should be reset instead
+ * @return the response from the REST API, if the request was successful
+ * @throws InterruptedException if the thread is interrupted while waiting for a
+ * request to modify the connector's offsets to succeed
+ * @see <a href="https://issues.apache.org/jira/browse/KAFKA-15826">KAFKA-15826</a>
+ */
+ private String modifySinkConnectorOffsetsWithRetry(ConnectorOffsets offsetsToAlter) throws InterruptedException {
+ // Some retry logic is necessary to account for KAFKA-15826,
+ // where laggy sink task startup/shutdown can leave consumers running
+ String conditionDetails = "Failed to "
+ + (offsetsToAlter != null ? "alter" : "reset")
+ + " sink connector offsets in time";
+ AtomicReference<String> responseReference = new AtomicReference<>();
+ waitForCondition(
+ () -> {
+ try {
+ if (offsetsToAlter == null) {
+ responseReference.set(connect.resetConnectorOffsets(connectorName));
+ } else {
+ responseReference.set(connect.alterConnectorOffsets(connectorName, offsetsToAlter));
+ }
+ return true;
+ } catch (ConnectRestException e) {
+ boolean internalServerError = e.statusCode() == INTERNAL_SERVER_ERROR.getStatusCode();
+
+ String message = Optional.of(e.getMessage()).orElse("");
+ boolean failedToResetConsumerOffsets = message.contains("Failed to reset consumer group offsets for connector");
Review Comment:
It looks like this method can also be used to alter a sink connector's offsets so shouldn't we account for that failure message as well?
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -895,6 +909,55 @@ private Map<String, String> baseSourceConnectorConfigs() {
return props;
}
+ /**
+ * Modify (i.e., alter or reset) the offsets for a sink connector, with retry logic to
+ * handle cases where laggy task shutdown may have left a consumer in the group.
+ * @param offsetsToAlter the offsets to alter for the sink connector, or null if
+ * the connector's offets should be reset instead
Review Comment:
```suggestion
* the connector's offsets should be reset instead
```
nit: typo
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
Posted by "yashmayya (via GitHub)" <gi...@apache.org>.
yashmayya commented on code in PR #14966:
URL: https://github.com/apache/kafka/pull/14966#discussion_r1426668693
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -895,6 +909,55 @@ private Map<String, String> baseSourceConnectorConfigs() {
return props;
}
+ /**
+ * Modify (i.e., alter or reset) the offsets for a sink connector, with retry logic to
+ * handle cases where laggy task shutdown may have left a consumer in the group.
+ * @param offsetsToAlter the offsets to alter for the sink connector, or null if
+ * the connector's offets should be reset instead
+ * @return the response from the REST API, if the request was successful
+ * @throws InterruptedException if the thread is interrupted while waiting for a
+ * request to modify the connector's offsets to succeed
+ * @see <a href="https://issues.apache.org/jira/browse/KAFKA-15826">KAFKA-15826</a>
+ */
+ private String modifySinkConnectorOffsetsWithRetry(ConnectorOffsets offsetsToAlter) throws InterruptedException {
+ // Some retry logic is necessary to account for KAFKA-15826,
+ // where laggy sink task startup/shutdown can leave consumers running
Review Comment:
Okay fair enough, thanks for the explanation!
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -895,6 +909,56 @@ private Map<String, String> baseSourceConnectorConfigs() {
return props;
}
+ /**
+ * Modify (i.e., alter or reset) the offsets for a sink connector, with retry logic to
+ * handle cases where laggy task shutdown may have left a consumer in the group.
+ * @param offsetsToAlter the offsets to alter for the sink connector, or null if
+ * the connector's offsets should be reset instead
+ * @return the response from the REST API, if the request was successful
+ * @throws InterruptedException if the thread is interrupted while waiting for a
+ * request to modify the connector's offsets to succeed
+ * @see <a href="https://issues.apache.org/jira/browse/KAFKA-15826">KAFKA-15826</a>
+ */
+ private String modifySinkConnectorOffsetsWithRetry(ConnectorOffsets offsetsToAlter) throws InterruptedException {
+ // Some retry logic is necessary to account for KAFKA-15826,
+ // where laggy sink task startup/shutdown can leave consumers running
+ String modifyVerb = offsetsToAlter != null ? "alter" : "reset";
+ String conditionDetails = "Failed to " + modifyVerb + " sink connector offsets in time";
+ AtomicReference<String> responseReference = new AtomicReference<>();
+ waitForCondition(
+ () -> {
+ try {
+ if (offsetsToAlter == null) {
+ responseReference.set(connect.resetConnectorOffsets(connectorName));
+ } else {
+ responseReference.set(connect.alterConnectorOffsets(connectorName, offsetsToAlter));
+ }
+ return true;
+ } catch (ConnectRestException e) {
+ boolean internalServerError = e.statusCode() == INTERNAL_SERVER_ERROR.getStatusCode();
+
+ String message = Optional.of(e.getMessage()).orElse("");
+ boolean failedToResetConsumerOffsets = message.contains(
Review Comment:
```suggestion
boolean failedToModifyConsumerOffsets = message.contains(
```
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on code in PR #14966:
URL: https://github.com/apache/kafka/pull/14966#discussion_r1421385777
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -324,127 +363,151 @@ public void testAlterSinkConnectorOffsetsDifferentKafkaClusterTargeted() throws
private void alterAndVerifySinkConnectorOffsets(Map<String, String> connectorConfigs, EmbeddedKafkaCluster kafkaCluster) throws Exception {
int numPartitions = 3;
- kafkaCluster.createTopic(TOPIC, numPartitions);
+ kafkaCluster.createTopic(topic, numPartitions);
// Produce records to each partition
for (int partition = 0; partition < numPartitions; partition++) {
for (int record = 0; record < NUM_RECORDS_PER_PARTITION; record++) {
- kafkaCluster.produce(TOPIC, partition, "key", "value");
+ kafkaCluster.produce(topic, partition, "key", "value");
}
}
// Create sink connector
- connect.configureConnector(CONNECTOR_NAME, connectorConfigs);
- connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(CONNECTOR_NAME, NUM_TASKS,
+ connect.configureConnector(connectorName, connectorConfigs);
+ connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(connectorName, NUM_TASKS,
"Connector tasks did not start in time.");
- verifyExpectedSinkConnectorOffsets(CONNECTOR_NAME, TOPIC, numPartitions, NUM_RECORDS_PER_PARTITION,
+ verifyExpectedSinkConnectorOffsets(connectorName, topic, numPartitions, NUM_RECORDS_PER_PARTITION,
"Sink connector consumer group offsets should catch up to the topic end offsets");
- connect.stopConnector(CONNECTOR_NAME);
+ connect.stopConnector(connectorName);
connect.assertions().assertConnectorIsStopped(
- CONNECTOR_NAME,
+ connectorName,
"Connector did not stop in time"
);
// Delete the offset of one partition; alter the offsets of the others
List<ConnectorOffset> offsetsToAlter = new ArrayList<>();
Map<String, Object> partition = new HashMap<>();
- partition.put(SinkUtils.KAFKA_TOPIC_KEY, TOPIC);
+ partition.put(SinkUtils.KAFKA_TOPIC_KEY, topic);
partition.put(SinkUtils.KAFKA_PARTITION_KEY, 0);
offsetsToAlter.add(new ConnectorOffset(partition, null));
for (int i = 1; i < numPartitions; i++) {
partition = new HashMap<>();
- partition.put(SinkUtils.KAFKA_TOPIC_KEY, TOPIC);
+ partition.put(SinkUtils.KAFKA_TOPIC_KEY, topic);
partition.put(SinkUtils.KAFKA_PARTITION_KEY, i);
offsetsToAlter.add(new ConnectorOffset(partition, Collections.singletonMap(SinkUtils.KAFKA_OFFSET_KEY, 5)));
}
- String response = connect.alterConnectorOffsets(CONNECTOR_NAME, new ConnectorOffsets(offsetsToAlter));
+ AtomicReference<String> responseReference = new AtomicReference<>();
+ // Some retry logic is necessary to account for KAFKA-15826,
+ // where laggy sink task startup/shutdown can leave consumers running
+ waitForCondition(
+ () -> {
+ try {
+ responseReference.set(connect.alterConnectorOffsets(connectorName, new ConnectorOffsets(offsetsToAlter)));
+ return true;
+ } catch (ConnectRestException e) {
+ boolean internalServerError = e.statusCode() == INTERNAL_SERVER_ERROR.getStatusCode();
+
+ String message = Optional.of(e.getMessage()).orElse("");
+ boolean failedToResetConsumerOffsets = message.contains("Failed to reset consumer group offsets for connector");
+ boolean canBeRetried = message.contains("If the connector is in a stopped state, this operation can be safely retried");
+
+ boolean retriable = internalServerError && failedToResetConsumerOffsets && canBeRetried;
+ if (retriable) {
+ return false;
+ } else {
+ throw new NoRetryException(e);
+ }
Review Comment:
nit, could be simplified to
```
if (retriable) {
return false;
}
throw new NoRetryException(e);
```
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -70,74 +80,115 @@ public class OffsetsApiIntegrationTest {
private static final long OFFSET_COMMIT_INTERVAL_MS = TimeUnit.SECONDS.toMillis(1);
private static final long OFFSET_READ_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(30);
private static final int NUM_WORKERS = 3;
- private static final String CONNECTOR_NAME = "test-connector";
- private static final String TOPIC = "test-topic";
private static final int NUM_TASKS = 2;
private static final int NUM_RECORDS_PER_PARTITION = 10;
- private Map<String, String> workerProps;
- private EmbeddedConnectCluster.Builder connectBuilder;
+ private static final Map<Map<String, String>, EmbeddedConnectCluster> CONNECT_CLUSTERS = new ConcurrentHashMap<>();
+ @Rule
+ public TestName currentTest = new TestName();
private EmbeddedConnectCluster connect;
+ private String connectorName;
+ private String topic;
@Before
public void setup() {
- Properties brokerProps = new Properties();
- brokerProps.put("transaction.state.log.replication.factor", "1");
- brokerProps.put("transaction.state.log.min.isr", "1");
-
- // setup Connect worker properties
- workerProps = new HashMap<>();
- workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
-
- // build a Connect cluster backed by Kafka and Zk
- connectBuilder = new EmbeddedConnectCluster.Builder()
- .name("connect-cluster")
- .numWorkers(NUM_WORKERS)
- .brokerProps(brokerProps)
- .workerProps(workerProps);
+ connectorName = currentTest.getMethodName();
+ topic = currentTest.getMethodName();
+ connect = defaultConnectCluster();
}
@After
public void tearDown() {
- connect.stop();
+ Set<String> remainingConnectors = new HashSet<>(connect.connectors());
+ if (remainingConnectors.remove(connectorName)) {
+ connect.deleteConnector(connectorName);
+ }
+ try {
+ assertEquals(
+ "Some connectors were not properly cleaned up after this test",
+ Collections.emptySet(),
+ remainingConnectors
+ );
+ } finally {
+ // Make a last-ditch effort to clean up the leaked connectors
+ // so as not to interfere with other test cases
+ remainingConnectors.forEach(connect::deleteConnector);
+ }
+ }
+
+ @AfterClass
+ public static void close() {
+ // stop all Connect, Kafka and Zk threads.
+ CONNECT_CLUSTERS.values().forEach(EmbeddedConnectCluster::stop);
+ }
+
+ private static EmbeddedConnectCluster createOrReuseConnectWithWorkerProps(Map<String, String> workerProps) {
+ return CONNECT_CLUSTERS.computeIfAbsent(workerProps, props -> {
+ Properties brokerProps = new Properties();
+ brokerProps.put("transaction.state.log.replication.factor", "1");
+ brokerProps.put("transaction.state.log.min.isr", "1");
+
+ // Have to declare a new map since the passed-in one may be immutable
+ Map<String, String> workerPropsWithDefaults = new HashMap<>(workerProps);
+ // Enable fast offset commits by default
+ workerPropsWithDefaults.putIfAbsent(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
+
+ EmbeddedConnectCluster result = new EmbeddedConnectCluster.Builder()
+ .name("connect-cluster")
+ .numWorkers(NUM_WORKERS)
+ .brokerProps(brokerProps)
+ .workerProps(workerPropsWithDefaults)
+ .build();
+
+ result.start();
+
+ return result;
+ });
+ }
+
+ private static EmbeddedConnectCluster defaultConnectCluster() {
+ return createOrReuseConnectWithWorkerProps(Collections.emptyMap());
+ }
+
+ private static EmbeddedConnectCluster exactlyOnceSourceConnectCluster() {
+ Map<String, String> workerProps = Collections.singletonMap(
+ DistributedConfig.EXACTLY_ONCE_SOURCE_SUPPORT_CONFIG,
+ "enabled"
+ );
+ return createOrReuseConnectWithWorkerProps(workerProps);
}
@Test
public void testGetNonExistentConnectorOffsets() {
- connect = connectBuilder.build();
- connect.start();
ConnectRestException e = assertThrows(ConnectRestException.class,
() -> connect.connectorOffsets("non-existent-connector"));
assertEquals(404, e.errorCode());
}
@Test
public void testGetSinkConnectorOffsets() throws Exception {
- connect = connectBuilder.build();
- connect.start();
getAndVerifySinkConnectorOffsets(baseSinkConnectorConfigs(), connect.kafka());
}
@Test
public void testGetSinkConnectorOffsetsOverriddenConsumerGroupId() throws Exception {
- connect = connectBuilder.build();
- connect.start();
Map<String, String> connectorConfigs = baseSinkConnectorConfigs();
- connectorConfigs.put(ConnectorConfig.CONNECTOR_CLIENT_CONSUMER_OVERRIDES_PREFIX + CommonClientConfigs.GROUP_ID_CONFIG,
- "overridden-group-id");
+ String overriddenGroupId = connectorName = "-overridden-group-id";
Review Comment:
nit: `overridden-group-id` could be a string constant at a class level? I see it being used multiple times. I mean it was the same in the prior version as well but anyways we are changing the class now so we can make this small change as well.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -70,74 +80,115 @@ public class OffsetsApiIntegrationTest {
private static final long OFFSET_COMMIT_INTERVAL_MS = TimeUnit.SECONDS.toMillis(1);
private static final long OFFSET_READ_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(30);
private static final int NUM_WORKERS = 3;
- private static final String CONNECTOR_NAME = "test-connector";
- private static final String TOPIC = "test-topic";
private static final int NUM_TASKS = 2;
private static final int NUM_RECORDS_PER_PARTITION = 10;
- private Map<String, String> workerProps;
- private EmbeddedConnectCluster.Builder connectBuilder;
+ private static final Map<Map<String, String>, EmbeddedConnectCluster> CONNECT_CLUSTERS = new ConcurrentHashMap<>();
+ @Rule
+ public TestName currentTest = new TestName();
private EmbeddedConnectCluster connect;
+ private String connectorName;
+ private String topic;
@Before
public void setup() {
- Properties brokerProps = new Properties();
- brokerProps.put("transaction.state.log.replication.factor", "1");
- brokerProps.put("transaction.state.log.min.isr", "1");
-
- // setup Connect worker properties
- workerProps = new HashMap<>();
- workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
-
- // build a Connect cluster backed by Kafka and Zk
- connectBuilder = new EmbeddedConnectCluster.Builder()
- .name("connect-cluster")
- .numWorkers(NUM_WORKERS)
- .brokerProps(brokerProps)
- .workerProps(workerProps);
+ connectorName = currentTest.getMethodName();
+ topic = currentTest.getMethodName();
+ connect = defaultConnectCluster();
}
@After
public void tearDown() {
- connect.stop();
+ Set<String> remainingConnectors = new HashSet<>(connect.connectors());
+ if (remainingConnectors.remove(connectorName)) {
+ connect.deleteConnector(connectorName);
+ }
+ try {
+ assertEquals(
+ "Some connectors were not properly cleaned up after this test",
+ Collections.emptySet(),
+ remainingConnectors
+ );
+ } finally {
+ // Make a last-ditch effort to clean up the leaked connectors
+ // so as not to interfere with other test cases
+ remainingConnectors.forEach(connect::deleteConnector);
+ }
+ }
+
+ @AfterClass
+ public static void close() {
+ // stop all Connect, Kafka and Zk threads.
+ CONNECT_CLUSTERS.values().forEach(EmbeddedConnectCluster::stop);
+ }
+
+ private static EmbeddedConnectCluster createOrReuseConnectWithWorkerProps(Map<String, String> workerProps) {
+ return CONNECT_CLUSTERS.computeIfAbsent(workerProps, props -> {
+ Properties brokerProps = new Properties();
+ brokerProps.put("transaction.state.log.replication.factor", "1");
+ brokerProps.put("transaction.state.log.min.isr", "1");
+
+ // Have to declare a new map since the passed-in one may be immutable
+ Map<String, String> workerPropsWithDefaults = new HashMap<>(workerProps);
+ // Enable fast offset commits by default
+ workerPropsWithDefaults.putIfAbsent(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
+
+ EmbeddedConnectCluster result = new EmbeddedConnectCluster.Builder()
Review Comment:
nit: We could call `result` as `connect` maybe ?(similar to the ConnectRestartAPI test above).
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -324,127 +363,151 @@ public void testAlterSinkConnectorOffsetsDifferentKafkaClusterTargeted() throws
private void alterAndVerifySinkConnectorOffsets(Map<String, String> connectorConfigs, EmbeddedKafkaCluster kafkaCluster) throws Exception {
int numPartitions = 3;
- kafkaCluster.createTopic(TOPIC, numPartitions);
+ kafkaCluster.createTopic(topic, numPartitions);
// Produce records to each partition
for (int partition = 0; partition < numPartitions; partition++) {
for (int record = 0; record < NUM_RECORDS_PER_PARTITION; record++) {
- kafkaCluster.produce(TOPIC, partition, "key", "value");
+ kafkaCluster.produce(topic, partition, "key", "value");
}
}
// Create sink connector
- connect.configureConnector(CONNECTOR_NAME, connectorConfigs);
- connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(CONNECTOR_NAME, NUM_TASKS,
+ connect.configureConnector(connectorName, connectorConfigs);
+ connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(connectorName, NUM_TASKS,
"Connector tasks did not start in time.");
- verifyExpectedSinkConnectorOffsets(CONNECTOR_NAME, TOPIC, numPartitions, NUM_RECORDS_PER_PARTITION,
+ verifyExpectedSinkConnectorOffsets(connectorName, topic, numPartitions, NUM_RECORDS_PER_PARTITION,
"Sink connector consumer group offsets should catch up to the topic end offsets");
- connect.stopConnector(CONNECTOR_NAME);
+ connect.stopConnector(connectorName);
connect.assertions().assertConnectorIsStopped(
- CONNECTOR_NAME,
+ connectorName,
"Connector did not stop in time"
);
// Delete the offset of one partition; alter the offsets of the others
List<ConnectorOffset> offsetsToAlter = new ArrayList<>();
Map<String, Object> partition = new HashMap<>();
- partition.put(SinkUtils.KAFKA_TOPIC_KEY, TOPIC);
+ partition.put(SinkUtils.KAFKA_TOPIC_KEY, topic);
partition.put(SinkUtils.KAFKA_PARTITION_KEY, 0);
offsetsToAlter.add(new ConnectorOffset(partition, null));
for (int i = 1; i < numPartitions; i++) {
partition = new HashMap<>();
- partition.put(SinkUtils.KAFKA_TOPIC_KEY, TOPIC);
+ partition.put(SinkUtils.KAFKA_TOPIC_KEY, topic);
partition.put(SinkUtils.KAFKA_PARTITION_KEY, i);
offsetsToAlter.add(new ConnectorOffset(partition, Collections.singletonMap(SinkUtils.KAFKA_OFFSET_KEY, 5)));
}
- String response = connect.alterConnectorOffsets(CONNECTOR_NAME, new ConnectorOffsets(offsetsToAlter));
+ AtomicReference<String> responseReference = new AtomicReference<>();
+ // Some retry logic is necessary to account for KAFKA-15826,
+ // where laggy sink task startup/shutdown can leave consumers running
+ waitForCondition(
+ () -> {
+ try {
+ responseReference.set(connect.alterConnectorOffsets(connectorName, new ConnectorOffsets(offsetsToAlter)));
+ return true;
+ } catch (ConnectRestException e) {
+ boolean internalServerError = e.statusCode() == INTERNAL_SERVER_ERROR.getStatusCode();
+
+ String message = Optional.of(e.getMessage()).orElse("");
+ boolean failedToResetConsumerOffsets = message.contains("Failed to reset consumer group offsets for connector");
+ boolean canBeRetried = message.contains("If the connector is in a stopped state, this operation can be safely retried");
+
+ boolean retriable = internalServerError && failedToResetConsumerOffsets && canBeRetried;
+ if (retriable) {
+ return false;
+ } else {
+ throw new NoRetryException(e);
+ }
+ } catch (Throwable t) {
+ throw new NoRetryException(t);
+ }
+ },
+ 30_000,
+ "Failed to reset sink connector offsets in time"
+ );
+ String response = responseReference.get();
Review Comment:
I see the same block of code in `resetAndVerifySinkConnectorOffsets` . We could extract it put in a method, call it something like `verifySinkConnectorOffsets` or something similar and invoke it?
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #14966:
URL: https://github.com/apache/kafka/pull/14966#discussion_r1423199242
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -70,74 +80,115 @@ public class OffsetsApiIntegrationTest {
private static final long OFFSET_COMMIT_INTERVAL_MS = TimeUnit.SECONDS.toMillis(1);
private static final long OFFSET_READ_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(30);
private static final int NUM_WORKERS = 3;
- private static final String CONNECTOR_NAME = "test-connector";
- private static final String TOPIC = "test-topic";
private static final int NUM_TASKS = 2;
private static final int NUM_RECORDS_PER_PARTITION = 10;
- private Map<String, String> workerProps;
- private EmbeddedConnectCluster.Builder connectBuilder;
+ private static final Map<Map<String, String>, EmbeddedConnectCluster> CONNECT_CLUSTERS = new ConcurrentHashMap<>();
+ @Rule
+ public TestName currentTest = new TestName();
private EmbeddedConnectCluster connect;
+ private String connectorName;
+ private String topic;
@Before
public void setup() {
- Properties brokerProps = new Properties();
- brokerProps.put("transaction.state.log.replication.factor", "1");
- brokerProps.put("transaction.state.log.min.isr", "1");
-
- // setup Connect worker properties
- workerProps = new HashMap<>();
- workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
-
- // build a Connect cluster backed by Kafka and Zk
- connectBuilder = new EmbeddedConnectCluster.Builder()
- .name("connect-cluster")
- .numWorkers(NUM_WORKERS)
- .brokerProps(brokerProps)
- .workerProps(workerProps);
+ connectorName = currentTest.getMethodName();
Review Comment:
As long as we actually use the `connectorName` field (which at this point is easier to use in testing code than a hardcoded constant) when doing `startsWith` checks, this should be fine.
It'd be fairly involved to try to wipe the offsets topic after every test (and this would only cover source connectors; it'd be more work for sink connectors too). We can't delete and recreate the topic without also restarting our embedded Connect workers, which would degrade the improvements from this PR. To be completely safe, we'd have to read the topic, find all source offsets present in it, and produce tombstones for all of them.
IMO it's just not worth the effort.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -70,74 +80,115 @@ public class OffsetsApiIntegrationTest {
private static final long OFFSET_COMMIT_INTERVAL_MS = TimeUnit.SECONDS.toMillis(1);
private static final long OFFSET_READ_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(30);
private static final int NUM_WORKERS = 3;
- private static final String CONNECTOR_NAME = "test-connector";
- private static final String TOPIC = "test-topic";
private static final int NUM_TASKS = 2;
private static final int NUM_RECORDS_PER_PARTITION = 10;
- private Map<String, String> workerProps;
- private EmbeddedConnectCluster.Builder connectBuilder;
+ private static final Map<Map<String, String>, EmbeddedConnectCluster> CONNECT_CLUSTERS = new ConcurrentHashMap<>();
+ @Rule
+ public TestName currentTest = new TestName();
private EmbeddedConnectCluster connect;
+ private String connectorName;
+ private String topic;
@Before
public void setup() {
- Properties brokerProps = new Properties();
- brokerProps.put("transaction.state.log.replication.factor", "1");
- brokerProps.put("transaction.state.log.min.isr", "1");
-
- // setup Connect worker properties
- workerProps = new HashMap<>();
- workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
-
- // build a Connect cluster backed by Kafka and Zk
- connectBuilder = new EmbeddedConnectCluster.Builder()
- .name("connect-cluster")
- .numWorkers(NUM_WORKERS)
- .brokerProps(brokerProps)
- .workerProps(workerProps);
+ connectorName = currentTest.getMethodName();
+ topic = currentTest.getMethodName();
+ connect = defaultConnectCluster();
}
@After
public void tearDown() {
- connect.stop();
+ Set<String> remainingConnectors = new HashSet<>(connect.connectors());
+ if (remainingConnectors.remove(connectorName)) {
+ connect.deleteConnector(connectorName);
+ }
+ try {
Review Comment:
IMO it's not worth the bother to execute another REST request just to make sure that deleting the connector went off as expected. We have plenty of test coverage for connector deletion logic elsewhere.
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -70,74 +80,115 @@ public class OffsetsApiIntegrationTest {
private static final long OFFSET_COMMIT_INTERVAL_MS = TimeUnit.SECONDS.toMillis(1);
private static final long OFFSET_READ_TIMEOUT_MS = TimeUnit.SECONDS.toMillis(30);
private static final int NUM_WORKERS = 3;
- private static final String CONNECTOR_NAME = "test-connector";
- private static final String TOPIC = "test-topic";
private static final int NUM_TASKS = 2;
private static final int NUM_RECORDS_PER_PARTITION = 10;
- private Map<String, String> workerProps;
- private EmbeddedConnectCluster.Builder connectBuilder;
+ private static final Map<Map<String, String>, EmbeddedConnectCluster> CONNECT_CLUSTERS = new ConcurrentHashMap<>();
+ @Rule
+ public TestName currentTest = new TestName();
private EmbeddedConnectCluster connect;
+ private String connectorName;
+ private String topic;
@Before
public void setup() {
- Properties brokerProps = new Properties();
- brokerProps.put("transaction.state.log.replication.factor", "1");
- brokerProps.put("transaction.state.log.min.isr", "1");
-
- // setup Connect worker properties
- workerProps = new HashMap<>();
- workerProps.put(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
-
- // build a Connect cluster backed by Kafka and Zk
- connectBuilder = new EmbeddedConnectCluster.Builder()
- .name("connect-cluster")
- .numWorkers(NUM_WORKERS)
- .brokerProps(brokerProps)
- .workerProps(workerProps);
+ connectorName = currentTest.getMethodName();
+ topic = currentTest.getMethodName();
+ connect = defaultConnectCluster();
}
@After
public void tearDown() {
- connect.stop();
+ Set<String> remainingConnectors = new HashSet<>(connect.connectors());
+ if (remainingConnectors.remove(connectorName)) {
+ connect.deleteConnector(connectorName);
+ }
+ try {
+ assertEquals(
+ "Some connectors were not properly cleaned up after this test",
+ Collections.emptySet(),
+ remainingConnectors
+ );
+ } finally {
+ // Make a last-ditch effort to clean up the leaked connectors
+ // so as not to interfere with other test cases
+ remainingConnectors.forEach(connect::deleteConnector);
+ }
+ }
+
+ @AfterClass
+ public static void close() {
+ // stop all Connect, Kafka and Zk threads.
+ CONNECT_CLUSTERS.values().forEach(EmbeddedConnectCluster::stop);
+ }
+
+ private static EmbeddedConnectCluster createOrReuseConnectWithWorkerProps(Map<String, String> workerProps) {
+ return CONNECT_CLUSTERS.computeIfAbsent(workerProps, props -> {
+ Properties brokerProps = new Properties();
+ brokerProps.put("transaction.state.log.replication.factor", "1");
+ brokerProps.put("transaction.state.log.min.isr", "1");
+
+ // Have to declare a new map since the passed-in one may be immutable
+ Map<String, String> workerPropsWithDefaults = new HashMap<>(workerProps);
+ // Enable fast offset commits by default
+ workerPropsWithDefaults.putIfAbsent(OFFSET_COMMIT_INTERVAL_MS_CONFIG, String.valueOf(OFFSET_COMMIT_INTERVAL_MS));
+
+ EmbeddedConnectCluster result = new EmbeddedConnectCluster.Builder()
Review Comment:
This is a very standard variable name for return values. I think it's clear enough (and to some degree, clearer) with the current name.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-15988: Reuse embedded clusters across test cases in Connect OffsetsApiIntegrationTest suite [kafka]
Posted by "C0urante (via GitHub)" <gi...@apache.org>.
C0urante commented on code in PR #14966:
URL: https://github.com/apache/kafka/pull/14966#discussion_r1423211506
##########
connect/runtime/src/test/java/org/apache/kafka/connect/integration/OffsetsApiIntegrationTest.java:
##########
@@ -324,127 +363,151 @@ public void testAlterSinkConnectorOffsetsDifferentKafkaClusterTargeted() throws
private void alterAndVerifySinkConnectorOffsets(Map<String, String> connectorConfigs, EmbeddedKafkaCluster kafkaCluster) throws Exception {
int numPartitions = 3;
- kafkaCluster.createTopic(TOPIC, numPartitions);
+ kafkaCluster.createTopic(topic, numPartitions);
// Produce records to each partition
for (int partition = 0; partition < numPartitions; partition++) {
for (int record = 0; record < NUM_RECORDS_PER_PARTITION; record++) {
- kafkaCluster.produce(TOPIC, partition, "key", "value");
+ kafkaCluster.produce(topic, partition, "key", "value");
}
}
// Create sink connector
- connect.configureConnector(CONNECTOR_NAME, connectorConfigs);
- connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(CONNECTOR_NAME, NUM_TASKS,
+ connect.configureConnector(connectorName, connectorConfigs);
+ connect.assertions().assertConnectorAndAtLeastNumTasksAreRunning(connectorName, NUM_TASKS,
"Connector tasks did not start in time.");
- verifyExpectedSinkConnectorOffsets(CONNECTOR_NAME, TOPIC, numPartitions, NUM_RECORDS_PER_PARTITION,
+ verifyExpectedSinkConnectorOffsets(connectorName, topic, numPartitions, NUM_RECORDS_PER_PARTITION,
"Sink connector consumer group offsets should catch up to the topic end offsets");
- connect.stopConnector(CONNECTOR_NAME);
+ connect.stopConnector(connectorName);
connect.assertions().assertConnectorIsStopped(
- CONNECTOR_NAME,
+ connectorName,
"Connector did not stop in time"
);
// Delete the offset of one partition; alter the offsets of the others
List<ConnectorOffset> offsetsToAlter = new ArrayList<>();
Map<String, Object> partition = new HashMap<>();
- partition.put(SinkUtils.KAFKA_TOPIC_KEY, TOPIC);
+ partition.put(SinkUtils.KAFKA_TOPIC_KEY, topic);
partition.put(SinkUtils.KAFKA_PARTITION_KEY, 0);
offsetsToAlter.add(new ConnectorOffset(partition, null));
for (int i = 1; i < numPartitions; i++) {
partition = new HashMap<>();
- partition.put(SinkUtils.KAFKA_TOPIC_KEY, TOPIC);
+ partition.put(SinkUtils.KAFKA_TOPIC_KEY, topic);
partition.put(SinkUtils.KAFKA_PARTITION_KEY, i);
offsetsToAlter.add(new ConnectorOffset(partition, Collections.singletonMap(SinkUtils.KAFKA_OFFSET_KEY, 5)));
}
- String response = connect.alterConnectorOffsets(CONNECTOR_NAME, new ConnectorOffsets(offsetsToAlter));
+ AtomicReference<String> responseReference = new AtomicReference<>();
+ // Some retry logic is necessary to account for KAFKA-15826,
+ // where laggy sink task startup/shutdown can leave consumers running
+ waitForCondition(
+ () -> {
+ try {
+ responseReference.set(connect.alterConnectorOffsets(connectorName, new ConnectorOffsets(offsetsToAlter)));
+ return true;
+ } catch (ConnectRestException e) {
+ boolean internalServerError = e.statusCode() == INTERNAL_SERVER_ERROR.getStatusCode();
+
+ String message = Optional.of(e.getMessage()).orElse("");
+ boolean failedToResetConsumerOffsets = message.contains("Failed to reset consumer group offsets for connector");
+ boolean canBeRetried = message.contains("If the connector is in a stopped state, this operation can be safely retried");
+
+ boolean retriable = internalServerError && failedToResetConsumerOffsets && canBeRetried;
+ if (retriable) {
+ return false;
+ } else {
+ throw new NoRetryException(e);
+ }
+ } catch (Throwable t) {
+ throw new NoRetryException(t);
+ }
+ },
+ 30_000,
+ "Failed to reset sink connector offsets in time"
+ );
+ String response = responseReference.get();
Review Comment:
Sure, done.
--
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.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org