mimaison commented on code in PR #23268:
URL: https://github.com/apache/kafka/pull/23268#discussion_r4221035057
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Herder.java:
##########
@@ -126,6 +126,25 @@ public interface Herder {
void putConnectorConfig(String connName, Map<String, String> config,
TargetState targetState, boolean allowReplace,
Callback<Created<ConnectorInfo>> callback);
+ /**
+ * Set the configuration for a connector, along with a target state and
initial offsets optionally. This supports
+ * creation and updating.
+ * @param connName name of the connector
+ * @param config the connector's configuration
+ * @param targetState the desired target state for the connector; may be
{@code null} if no target state change is desired. Note that the default
+ * target state is {@link TargetState#STARTED} if no
target state exists previously
+ * @param initialOffsets the offsets the connector should start from; may
be {@code null}, in which case the connector's
+ * existing offsets (if any) are left untouched.
When non-null, every offset the connector currently
+ * has is wiped before these are written, so the
connector ends up with exactly these offsets rather
+ * than a merge of old and new. Only meaningful when
creating a connector.
+ * @param allowReplace if true, allow overwriting previous configs; if
false, throw {@link AlreadyExistsException}
+ * if a connector with the same name already exists
+ * @param callback callback to invoke when the configuration has been
written
+ */
+ void putConnectorConfig(String connName, Map<String, String> config,
TargetState targetState,
Review Comment:
I wonder if we could drop the overload just above, it seems it's now only
called from tests.
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/AbstractHerder.java:
##########
@@ -1316,6 +1316,66 @@ public void resetConnectorOffsets(String connName,
Callback<Message> callback) {
*/
protected abstract void modifyConnectorOffsets(String connName,
Map<Map<String, ?>, Map<String, ?>> offsets, Callback<Message> cb);
+ /**
+ * Validate the offsets supplied via the {@code initial_offsets} field of
a connector creation request. Performs no
+ * I/O, so that an invalid request is rejected before any of the
connector's existing offsets are wiped.
+ *
+ * @param connName the name of the connector being created
+ * @param initialOffsets the offsets supplied by the user; may not be
{@code null}
+ * @throws BadRequestException if the offsets cannot be used as a
connector's initial offsets
+ */
+ protected void validateInitialOffsets(String connName, Map<Map<String, ?>,
Map<String, ?>> initialOffsets) {
+ if (initialOffsets.isEmpty()) {
+ throw new BadRequestException("The initial offsets for connector "
+ connName + " may not be empty");
+ }
+ // A null offset means "delete this partition's offset" when altering,
but is meaningless on create: every offset
+ // the connector has is wiped before these are written, so there is
nothing left to delete. Issuing the delete
+ // anyway targets the consumer group our own wipe just removed, which
the alter path (unlike the reset path) does
+ // not tolerate. This is deliberately stricter than PATCH
/connectors/{connector}/offsets.
+ // Not Map::containsValue, which throws rather than returning false on
the immutable maps Map.of(...) produces.
+ if (initialOffsets.values().stream().anyMatch(offset -> offset ==
null)) {
+ throw new BadRequestException("The initial offsets for connector "
+ connName + " may not contain null offset "
+ + "values. All existing offsets for the connector are
removed before the initial offsets are written, "
+ + "so a null offset has nothing to delete.");
+ }
+ }
+
+ /**
+ * Wipe every offset the connector currently has and write {@code
initialOffsets} in their place.
+ * <p>
+ * Unlike {@link #alterConnectorOffsets} and {@link
#resetConnectorOffsets}, this skips the usual preconditions (that
+ * the connector exists and is STOPPED with no running tasks): the
connector is being created, so it is not expected
+ * to exist and has no tasks. Its config has not been written yet, so
nothing can be running under this name.
+ */
+ protected void setInitialConnectorOffsets(String connName, Map<String,
String> config,
+ Map<Map<String, ?>, Map<String,
?>> initialOffsets, Callback<Message> cb) {
+ worker.modifyConnectorOffsets(connName, config, initialOffsets, true,
cb);
+ }
+
+ /**
+ * Remove the initial offsets written for a connector whose config could
not subsequently be written, so that a
+ * failed creation does not leave offsets behind for a connector that does
not exist, and then complete
+ * {@code callback} with the original failure rather than with any error
from the cleanup itself.
+ */
+ protected void wipeInitialOffsetsAfterFailedCreate(String connName,
Map<String, String> config,
+ Throwable
configWriteError, Callback<Created<ConnectorInfo>> callback) {
+ log.error("Failed to write the configuration for connector {} after
its initial offsets had been written; "
Review Comment:
Would "Failed to create ..." be more accurate? We can end up here when
failing validation
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java:
##########
@@ -1418,7 +1439,13 @@ void modifySinkConnectorOffsets(String connName,
Connector connector, Map<String
try {
Map<TopicPartition, Long> offsetsToWrite;
- if (isReset) {
+ // Both a reset and a replace start by tombstoning every
offset the consumer group currently has;
+ // a replace then overlays the requested offsets on top of
those tombstones.
+ if (isReset || replaceAllOffsets) {
+ String modification = isReset ? "resetting" :
"replacing";
Review Comment:
We can move this into the catch block below where it's used
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResourceTest.java:
##########
@@ -294,7 +294,7 @@ public void testCreateConnector() throws Throwable {
final ArgumentCaptor<Callback<Herder.Created<ConnectorInfo>>> cb =
ArgumentCaptor.forClass(Callback.class);
Review Comment:
Would it make sense to add a test for create with offsets here?
##########
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java:
##########
@@ -1146,33 +1155,112 @@ private void doPutConnectorConfig(
}
log.trace("Handling connector config request {}",
connName);
- if (!isLeader()) {
- callback.onCompletion(new NotLeaderException("Only
the leader can set connector configs.", leaderUrl()), null);
- return null;
- }
- boolean exists = configState.contains(connName);
- if (!allowReplace && exists) {
- callback.onCompletion(new
AlreadyExistsException("Connector " + connName + " already exists"), null);
+ if (!putConnectorConfigChecks(connName, allowReplace,
callback)) {
return null;
}
- log.trace("Submitting connector config {} {} {}",
connName, allowReplace, configState.connectors());
- writeToConfigTopicAsLeader(
- "writing a config for connector " + connName +
" to the config topic",
- () ->
configBackingStore.putConnectorConfig(connName, config, targetState)
- );
-
- // Note that we use the updated connector config
despite the fact that we don't have an updated
- // snapshot yet. The existing task info should still
be accurate.
- ConnectorInfo info = new ConnectorInfo(connName,
config, configState.tasks(connName),
- connectorType(config));
- callback.onCompletion(null, new Created<>(!exists,
info));
+ if (initialOffsets == null) {
+ writeConnectorConfigAndComplete(connName, config,
targetState, allowReplace, null, callback);
+ } else {
+ setInitialOffsetsThenWriteConfig(connName, config,
targetState, initialOffsets, allowReplace, callback);
+ }
return null;
},
forwardErrorAndTickThreadStages(callback)
);
}));
}
+
+ /**
+ * Perform the checks required before writing a connector's config as the
leader, completing {@code callback}
+ * exceptionally if any of them fail.
+ *
+ * @param connName the name of the connector whose config is to be written
+ * @param allowReplace whether an existing connector with this name may be
overwritten
+ * @param callback callback to complete exceptionally if a check fails
+ * @return true if all the checks passed, false otherwise
+ */
+ private boolean putConnectorConfigChecks(String connName, boolean
allowReplace, Callback<Created<ConnectorInfo>> callback) {
+ if (!isLeader()) {
+ callback.onCompletion(new NotLeaderException("Only the leader can
set connector configs.", leaderUrl()), null);
+ return false;
+ }
+ if (!allowReplace && configState.contains(connName)) {
+ callback.onCompletion(new AlreadyExistsException("Connector " +
connName + " already exists"), null);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * Write a connector's initial offsets and, once they are durable, write
its config. Must be invoked on the tick thread.
+ * <p>
+ * The offsets are written asynchronously, so the config write is handed
back to the tick thread via a nested request
+ * rather than blocking it; this mirrors the {@link
#modifyConnectorOffsets} plus zombie fencing flow, including the
+ * re-check of preconditions that may have changed while the offsets were
being written.
+ */
+ private void setInitialOffsetsThenWriteConfig(String connName, Map<String,
String> config, TargetState targetState,
+ Map<Map<String, ?>,
Map<String, ?>> initialOffsets, boolean allowReplace,
+
Callback<Created<ConnectorInfo>> callback) {
+ // Validate before anything destructive happens, so an invalid request
can't wipe the connector's offsets. This
+ // performs no I/O and is safe to run on the tick thread; throwing
here is forwarded to the callback by the
+ // enclosing herder request.
+ validateInitialOffsets(connName, initialOffsets);
+
+ try (TickThreadStage stage = new TickThreadStage("setting initial
offsets for connector " + connName)) {
+ setInitialConnectorOffsets(connName, config, initialOffsets,
(offsetsError, message) -> {
+ if (offsetsError != null) {
+ callback.onCompletion(offsetsError, null);
+ return;
+ }
+ addRequest(
+ () -> {
+ // The offsets are already durable, so wipe them back
out if the create can't be completed,
+ // whether because the precondition re-check fails
here or the config write below throws.
+ Callback<Created<ConnectorInfo>> wipeOnFailure =
(error, ignored) ->
+ wipeInitialOffsetsAfterFailedCreate(connName,
config, error, callback);
+ if (!putConnectorConfigChecks(connName, allowReplace,
wipeOnFailure)) {
+ return null;
+ }
+ try {
+ writeConnectorConfigAndComplete(connName, config,
targetState, allowReplace, message.message(), callback);
+ } catch (Throwable t) {
+ wipeInitialOffsetsAfterFailedCreate(connName,
config, t, callback);
+ }
+ return null;
+ },
+ forwardErrorAndTickThreadStages(callback)
+ );
+ });
+ }
+ }
+
+ /**
+ * Write a connector's config to the config topic and complete {@code
callback} with the resulting
+ * {@link ConnectorInfo}. Must be invoked on the tick thread.
+ *
+ * @param offsetsStatus the outcome of writing the connector's initial
offsets, to be reported back on the created
+ * {@link ConnectorInfo}; {@code null} if the request
did not supply initial offsets, in which
+ * case the response is identical to one for a
creation without initial offsets
+ */
+ private void writeConnectorConfigAndComplete(String connName, Map<String,
String> config, TargetState targetState,
Review Comment:
`allowReplace` is only used for a trace log line. Is it really helpful?
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/WorkerTest.java:
##########
@@ -2633,6 +2634,276 @@ public void
testResetOffsetsSinkConnectorDeleteConsumerGroupError(boolean enable
verifyKafkaClusterId();
}
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ @SuppressWarnings("unchecked")
+ public void testReplaceOffsetsSourceConnectorWipesStalePartitions(boolean
enableTopicCreation) throws Exception {
+ setup(enableTopicCreation);
+ mockKafkaClusterId();
+ mockInternalConverters();
+ worker = new Worker(WORKER_ID, new MockTime(), plugins, config,
offsetBackingStore, Executors.newSingleThreadExecutor(),
+ allConnectorClientConfigOverridePolicy, null);
+ worker.start();
+
+ when(plugins.withClassLoader(any(ClassLoader.class),
any(Runnable.class))).thenAnswer(AdditionalAnswers.returnsSecondArg());
+ when(sourceConnector.alterOffsets(eq(connectorProps),
anyMap())).thenReturn(true);
+ ConnectorOffsetBackingStore offsetStore =
mock(ConnectorOffsetBackingStore.class);
+ KafkaProducer<byte[], byte[]> producer = mock(KafkaProducer.class);
+ OffsetStorageWriter offsetWriter = mock(OffsetStorageWriter.class);
+
+ Map<String, Object> p1 = Map.of("partitionKey", "p1");
+ Map<String, Object> p2 = Map.of("partitionKey", "p2");
+ Map<String, Object> p3 = Map.of("partitionKey", "p3");
+ Set<Map<String, Object>> existingPartitions = new HashSet<>();
+ existingPartitions.add(p1);
+ existingPartitions.add(p2);
+
when(offsetStore.connectorPartitions(eq(CONNECTOR_ID))).thenReturn(existingPartitions);
+
+ // p2 already existed and is being re-requested with a new value; p3
is brand new; p1 existed but isn't
+ // requested, so it should be wiped rather than left behind
+ Map<Map<String, ?>, Map<String, ?>> requestedOffsets = new HashMap<>();
+ requestedOffsets.put(p2, Map.of("offsetKey", "newP2"));
+ requestedOffsets.put(p3, Map.of("offsetKey", "p3"));
+
+ when(offsetWriter.doFlush(any())).thenAnswer(invocation -> {
+ invocation.getArgument(0, Callback.class).onCompletion(null, null);
+ return null;
+ });
+
+ FutureCallback<Message> cb = new FutureCallback<>();
+ worker.modifySourceConnectorOffsets(CONNECTOR_ID, sourceConnector,
connectorProps, requestedOffsets, true, offsetStore, producer,
+ offsetWriter, Thread.currentThread().getContextClassLoader(),
cb);
+ assertEquals("The offsets for this connector have been set
successfully", cb.get(1000, TimeUnit.MILLISECONDS).message());
+
+ verify(offsetWriter).offset(p1, null);
+ verify(offsetWriter).offset(p2, Map.of("offsetKey", "newP2"));
+ verify(offsetWriter).offset(p3, Map.of("offsetKey", "p3"));
+ verify(offsetWriter, times(3)).offset(any(), any());
+ verify(offsetWriter).beginFlush();
+ verify(offsetStore, timeout(1000)).stop();
+ verifyKafkaClusterId();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ @SuppressWarnings("unchecked")
+ public void testReplaceOffsetsSourceConnectorNoPreexistingOffsets(boolean
enableTopicCreation) throws Exception {
+ setup(enableTopicCreation);
+ mockKafkaClusterId();
+ mockInternalConverters();
+ worker = new Worker(WORKER_ID, new MockTime(), plugins, config,
offsetBackingStore, Executors.newSingleThreadExecutor(),
+ allConnectorClientConfigOverridePolicy, null);
+ worker.start();
+
+ when(plugins.withClassLoader(any(ClassLoader.class),
any(Runnable.class))).thenAnswer(AdditionalAnswers.returnsSecondArg());
+ when(sourceConnector.alterOffsets(eq(connectorProps),
anyMap())).thenReturn(true);
+ ConnectorOffsetBackingStore offsetStore =
mock(ConnectorOffsetBackingStore.class);
+ KafkaProducer<byte[], byte[]> producer = mock(KafkaProducer.class);
+ OffsetStorageWriter offsetWriter = mock(OffsetStorageWriter.class);
+
+ // No pre-existing offsets for this connector -- there's nothing to
wipe, only something to write
+
when(offsetStore.connectorPartitions(eq(CONNECTOR_ID))).thenReturn(Set.of());
+
+ Map<Map<String, ?>, Map<String, ?>> requestedOffsets = Map.of(
+ Map.of("partitionKey", "partitionValue"), Map.of("offsetKey",
"offsetValue"));
+
+ when(offsetWriter.doFlush(any())).thenAnswer(invocation -> {
+ invocation.getArgument(0, Callback.class).onCompletion(null, null);
+ return null;
+ });
+
+ FutureCallback<Message> cb = new FutureCallback<>();
+ worker.modifySourceConnectorOffsets(CONNECTOR_ID, sourceConnector,
connectorProps, requestedOffsets, true, offsetStore, producer,
+ offsetWriter, Thread.currentThread().getContextClassLoader(),
cb);
+ assertEquals("The offsets for this connector have been set
successfully", cb.get(1000, TimeUnit.MILLISECONDS).message());
+
+ // The write must still happen even though there was nothing to wipe
-- guards against the "no offsets to
+ // write" shortcut (meant for a reset of a connector with no committed
offsets) swallowing a genuine write.
+ verify(offsetWriter).offset(Map.of("partitionKey", "partitionValue"),
Map.of("offsetKey", "offsetValue"));
+ verify(offsetWriter).beginFlush();
+ verify(offsetWriter).doFlush(any());
+ verify(offsetStore, timeout(1000)).stop();
+ verifyKafkaClusterId();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testReplaceOffsetsSinkConnectorWipesStalePartitions(boolean
enableTopicCreation) throws Exception {
+ setup(enableTopicCreation);
+ mockKafkaClusterId();
+ String connectorClass = SampleSinkConnector.class.getName();
+ connectorProps.put(CONNECTOR_CLASS_CONFIG, connectorClass);
+
+ Admin admin = mock(Admin.class);
+ worker = new Worker(WORKER_ID, new MockTime(), plugins, config,
offsetBackingStore, Executors.newCachedThreadPool(),
+ allConnectorClientConfigOverridePolicy, config -> admin);
+ worker.start();
+
+ when(plugins.withClassLoader(any(ClassLoader.class),
any(Runnable.class))).thenAnswer(AdditionalAnswers.returnsSecondArg());
+ when(sinkConnector.alterOffsets(eq(connectorProps),
anyMap())).thenReturn(true);
+
+ TopicPartition p1 = new TopicPartition("test_topic", 1);
+ TopicPartition p2 = new TopicPartition("test_topic", 2);
+ TopicPartition p3 = new TopicPartition("test_topic", 3);
+ mockAdminListConsumerGroupOffsets(admin, Map.of(p1, new
OffsetAndMetadata(10L), p2, new OffsetAndMetadata(20L)), null);
+
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor<Map<TopicPartition, OffsetAndMetadata>>
alterOffsetsMapCapture = ArgumentCaptor.forClass(Map.class);
+ AlterConsumerGroupOffsetsResult alterConsumerGroupOffsetsResult =
mock(AlterConsumerGroupOffsetsResult.class);
+ when(admin.alterConsumerGroupOffsets(anyString(),
alterOffsetsMapCapture.capture(), any(AlterConsumerGroupOffsetsOptions.class)))
+ .thenReturn(alterConsumerGroupOffsetsResult);
+
when(alterConsumerGroupOffsetsResult.all()).thenReturn(KafkaFuture.completedFuture(null));
+
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor<Set<TopicPartition>> deleteOffsetsSetCapture =
ArgumentCaptor.forClass(Set.class);
+ DeleteConsumerGroupOffsetsResult deleteConsumerGroupOffsetsResult =
mock(DeleteConsumerGroupOffsetsResult.class);
+ when(admin.deleteConsumerGroupOffsets(anyString(),
deleteOffsetsSetCapture.capture(),
any(DeleteConsumerGroupOffsetsOptions.class)))
+ .thenReturn(deleteConsumerGroupOffsetsResult);
+
when(deleteConsumerGroupOffsetsResult.all()).thenReturn(KafkaFuture.completedFuture(null));
+
+ // p2 already existed and is being re-requested with a new value; p3
is brand new; p1 existed but isn't
+ // requested, so it should be wiped rather than left behind
+ Map<String, String> partitionMap2 = new HashMap<>();
+ partitionMap2.put(SinkUtils.KAFKA_TOPIC_KEY, "test_topic");
+ partitionMap2.put(SinkUtils.KAFKA_PARTITION_KEY, "2");
+ Map<String, String> partitionMap3 = new HashMap<>();
+ partitionMap3.put(SinkUtils.KAFKA_TOPIC_KEY, "test_topic");
+ partitionMap3.put(SinkUtils.KAFKA_PARTITION_KEY, "3");
+ Map<Map<String, ?>, Map<String, ?>> requestedOffsets = new HashMap<>();
+ requestedOffsets.put(partitionMap2, Map.of(SinkUtils.KAFKA_OFFSET_KEY,
99));
+ requestedOffsets.put(partitionMap3, Map.of(SinkUtils.KAFKA_OFFSET_KEY,
5));
+
+ FutureCallback<Message> cb = new FutureCallback<>();
+ worker.modifySinkConnectorOffsets(CONNECTOR_ID, sinkConnector,
connectorProps, requestedOffsets, true,
+ Thread.currentThread().getContextClassLoader(), cb);
+ assertEquals("The offsets for this connector have been set
successfully", cb.get(1000, TimeUnit.MILLISECONDS).message());
+
+ assertEquals(Set.of(p1), deleteOffsetsSetCapture.getValue());
+ assertEquals(2, alterOffsetsMapCapture.getValue().size());
+ assertEquals(99, alterOffsetsMapCapture.getValue().get(p2).offset());
+ assertEquals(5, alterOffsetsMapCapture.getValue().get(p3).offset());
+
+ verify(admin, timeout(1000)).close();
+ verifyKafkaClusterId();
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {true, false})
+ public void testReplaceOffsetsSinkConnectorNoPreexistingOffsets(boolean
enableTopicCreation) throws Exception {
+ setup(enableTopicCreation);
+ mockKafkaClusterId();
+ String connectorClass = SampleSinkConnector.class.getName();
+ connectorProps.put(CONNECTOR_CLASS_CONFIG, connectorClass);
+
+ Admin admin = mock(Admin.class);
+ worker = new Worker(WORKER_ID, new MockTime(), plugins, config,
offsetBackingStore, Executors.newCachedThreadPool(),
+ allConnectorClientConfigOverridePolicy, config -> admin);
+ worker.start();
+
+ when(plugins.withClassLoader(any(ClassLoader.class),
any(Runnable.class))).thenAnswer(AdditionalAnswers.returnsSecondArg());
+ when(sinkConnector.alterOffsets(eq(connectorProps),
anyMap())).thenReturn(true);
+
+ // No pre-existing offsets for this consumer group -- there's nothing
to wipe, only something to write
+ mockAdminListConsumerGroupOffsets(admin, Map.of(), null);
+
+ @SuppressWarnings("unchecked")
+ ArgumentCaptor<Map<TopicPartition, OffsetAndMetadata>>
alterOffsetsMapCapture = ArgumentCaptor.forClass(Map.class);
+ AlterConsumerGroupOffsetsResult alterConsumerGroupOffsetsResult =
mock(AlterConsumerGroupOffsetsResult.class);
+ when(admin.alterConsumerGroupOffsets(anyString(),
alterOffsetsMapCapture.capture(), any(AlterConsumerGroupOffsetsOptions.class)))
+ .thenReturn(alterConsumerGroupOffsetsResult);
+
when(alterConsumerGroupOffsetsResult.all()).thenReturn(KafkaFuture.completedFuture(null));
+
+ TopicPartition p1 = new TopicPartition("test_topic", 1);
+ Map<String, String> partitionMap1 = new HashMap<>();
+ partitionMap1.put(SinkUtils.KAFKA_TOPIC_KEY, "test_topic");
+ partitionMap1.put(SinkUtils.KAFKA_PARTITION_KEY, "1");
+ Map<Map<String, ?>, Map<String, ?>> requestedOffsets =
Map.of(partitionMap1, Map.of(SinkUtils.KAFKA_OFFSET_KEY, 7));
+
+ FutureCallback<Message> cb = new FutureCallback<>();
+ worker.modifySinkConnectorOffsets(CONNECTOR_ID, sinkConnector,
connectorProps, requestedOffsets, true,
+ Thread.currentThread().getContextClassLoader(), cb);
+ assertEquals("The offsets for this connector have been set
successfully", cb.get(1000, TimeUnit.MILLISECONDS).message());
+
+ assertEquals(1, alterOffsetsMapCapture.getValue().size());
+ assertEquals(7, alterOffsetsMapCapture.getValue().get(p1).offset());
+ verify(admin, times(0)).deleteConsumerGroupOffsets(anyString(),
anySet(), any(DeleteConsumerGroupOffsetsOptions.class));
Review Comment:
`never()` is preferred over `times(0)`
##########
connect/runtime/src/test/java/org/apache/kafka/connect/runtime/standalone/StandaloneHerderTest.java:
##########
@@ -266,6 +270,56 @@ public void testCreateConnectorWithStoppedInitialState()
throws Exception {
verify(loaderSwap).close();
}
+ @Test
Review Comment:
In `DistributedHerderTest` we check the error paths, it seems we only check
the golden path here
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
