You are viewing a plain text version of this content. The canonical link for it is here.
Posted to pr@cassandra.apache.org by "5 (via GitHub)" <gi...@apache.org> on 2025/02/19 02:11:30 UTC
[PR] Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
5 opened a new pull request, #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198
https://issues.apache.org/jira/browse/CASSSIDECAR-203
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1987945295
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/ApiEndpointsV1.java:
##########
@@ -121,13 +121,16 @@ public final class ApiEndpointsV1
public static final String LIST_CDC_SEGMENTS_ROUTE = API_V1 + CDC_PATH + "/segments";
public static final String STREAM_CDC_SEGMENTS_ROUTE = LIST_CDC_SEGMENTS_ROUTE + "/" + SEGMENT_PATH_PARAM;
+ // Schema Reporting
+ private static final String REPORT_SCHEMA = "/report-schema";
+ public static final String REPORT_SCHEMA_ROUTE = API_V1 + REPORT_SCHEMA;
Review Comment:
I would agree that `PUT .../schema` (singular, not plural) is an appropriate name if we provide the schema we intent to persist. But we're not putting schema in. We're triggering an event that causes the schema to be internally produced and reported. Using `PUT .../schema-report` would probably follow the general idea of REST better. What do you think?
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1987939424
##########
server/src/test/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandlerTest.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import com.datastax.driver.core.Metadata;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Provides;
+import com.google.inject.Singleton;
+import com.google.inject.util.Modules;
+import com.linkedin.mxe.MetadataChangeProposal;
+import datahub.client.Callback;
+import datahub.client.MetadataWriteResponse;
+import datahub.shaded.findbugs.annotations.SuppressFBWarnings;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.Vertx;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.client.predicate.ResponsePredicate;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+import org.apache.cassandra.sidecar.TestModule;
+import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
+import org.apache.cassandra.sidecar.cluster.InstancesMetadata;
+import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.utils.IOUtils;
+import org.apache.cassandra.sidecar.datahub.EmitterFactory;
+import org.apache.cassandra.sidecar.datahub.JsonEmitter;
+import org.apache.cassandra.sidecar.server.MainModule;
+import org.apache.cassandra.sidecar.server.Server;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link ReportSchemaHandler}
+ */
+@ExtendWith(VertxExtension.class)
+final class ReportSchemaHandlerTest
+{
+ private static final String CLUSTER = "cluster";
+ private static final String DIRECTORY = "/tmp";
+ private static final int IDENTIFIER = 42;
+ private static final String LOCALHOST = "127.0.0.1";
+ private static final int PORT = 9042;
+ private static final String ENDPOINT = "/api/v1/report-schema";
+ private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+ private static final class ThrowingEmitter extends JsonEmitter
+ {
+ @Override
+ @NotNull
+ public synchronized Future<MetadataWriteResponse> emit(@NotNull MetadataChangeProposal proposal,
+ @Nullable Callback callback) throws IOException
+ {
+ throw new IOException();
+ }
+ }
+
+ private final class ReportSchemaHandlerTestModule extends AbstractModule
+ {
+ @Provides
+ @Singleton
+ @NotNull
+ public InstancesMetadata instancesMetadata()
+ {
+ Metadata metadata = mock(Metadata.class);
+ when(metadata.getKeyspaces()).thenReturn(Collections.emptyList());
+
+ StorageOperations operations = mock(StorageOperations.class);
+ when(operations.clusterName()).thenReturn(CLUSTER);
+
+ CassandraAdapterDelegate delegate = mock(CassandraAdapterDelegate.class);
+ when(delegate.storageOperations()).thenReturn(operations);
+ when(delegate.metadata()).thenReturn(metadata);
+
+ InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
+ when(instanceMetadata.stagingDir()).thenReturn(DIRECTORY);
+ when(instanceMetadata.id()).thenReturn(IDENTIFIER);
+ when(instanceMetadata.host()).thenReturn(LOCALHOST);
+ when(instanceMetadata.port()).thenReturn(PORT);
+ when(instanceMetadata.delegate()).thenReturn(delegate);
+
+ InstancesMetadata instances = mock(InstancesMetadata.class);
+ when(instances.instances()).thenReturn(Collections.singletonList(instanceMetadata));
+ when(instances.instanceFromId(IDENTIFIER)).thenReturn(instanceMetadata);
+ when(instances.instanceFromHost(LOCALHOST)).thenReturn(instanceMetadata);
+ return instances;
+ }
+
+ @Provides
+ @Singleton
+ @NotNull
+ public EmitterFactory emitterFactory()
+ {
+ return () -> emitter;
+ }
+ }
+
+ private final Injector injector = Guice.createInjector(Modules.override(new MainModule()).with(
+ Modules.override(new TestModule()).with(
+ new ReportSchemaHandlerTestModule())));
+ private WebClient client;
+ private Server server;
+ private JsonEmitter emitter;
+
+ @BeforeEach
+ void before() throws InterruptedException
+ {
+ VertxTestContext context = new VertxTestContext();
+ server = injector.getInstance(Server.class);
+ server.start()
+ .onSuccess(result -> context.completeNow())
+ .onFailure(context::failNow);
+
+ client = WebClient.create(injector.getInstance(Vertx.class));
+
+ context.awaitCompletion(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @AfterEach
+ @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED")
+ void after() throws InterruptedException
+ {
+ CountDownLatch latch = new CountDownLatch(1);
+ server.close()
+ .onSuccess(future -> latch.countDown());
+
+ client.close();
+
+ latch.await(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
Review Comment:
Replaced, 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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1984016599
##########
server/src/test/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandlerTest.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import com.datastax.driver.core.Metadata;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Provides;
+import com.google.inject.Singleton;
+import com.google.inject.util.Modules;
+import com.linkedin.mxe.MetadataChangeProposal;
+import datahub.client.Callback;
+import datahub.client.MetadataWriteResponse;
+import datahub.shaded.findbugs.annotations.SuppressFBWarnings;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.Vertx;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.client.predicate.ResponsePredicate;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+import org.apache.cassandra.sidecar.TestModule;
+import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
+import org.apache.cassandra.sidecar.cluster.InstancesMetadata;
+import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.utils.IOUtils;
+import org.apache.cassandra.sidecar.datahub.EmitterFactory;
+import org.apache.cassandra.sidecar.datahub.JsonEmitter;
+import org.apache.cassandra.sidecar.server.MainModule;
+import org.apache.cassandra.sidecar.server.Server;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link ReportSchemaHandler}
+ */
+@ExtendWith(VertxExtension.class)
+final class ReportSchemaHandlerTest
+{
+ private static final String CLUSTER = "cluster";
+ private static final String DIRECTORY = "/tmp";
+ private static final int IDENTIFIER = 42;
+ private static final String LOCALHOST = "127.0.0.1";
+ private static final int PORT = 9042;
+ private static final String ENDPOINT = "/api/v1/report-schema";
+ private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+ private static final class ThrowingEmitter extends JsonEmitter
+ {
+ @Override
+ @NotNull
+ public synchronized Future<MetadataWriteResponse> emit(@NotNull MetadataChangeProposal proposal,
+ @Nullable Callback callback) throws IOException
+ {
+ throw new IOException();
+ }
+ }
+
+ private final class ReportSchemaHandlerTestModule extends AbstractModule
+ {
+ @Provides
+ @Singleton
+ @NotNull
+ public InstancesMetadata instancesMetadata()
+ {
+ Metadata metadata = mock(Metadata.class);
+ when(metadata.getKeyspaces()).thenReturn(Collections.emptyList());
+
+ StorageOperations operations = mock(StorageOperations.class);
+ when(operations.clusterName()).thenReturn(CLUSTER);
+
+ CassandraAdapterDelegate delegate = mock(CassandraAdapterDelegate.class);
+ when(delegate.storageOperations()).thenReturn(operations);
+ when(delegate.metadata()).thenReturn(metadata);
+
+ InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
+ when(instanceMetadata.stagingDir()).thenReturn(DIRECTORY);
+ when(instanceMetadata.id()).thenReturn(IDENTIFIER);
+ when(instanceMetadata.host()).thenReturn(LOCALHOST);
+ when(instanceMetadata.port()).thenReturn(PORT);
+ when(instanceMetadata.delegate()).thenReturn(delegate);
+
+ InstancesMetadata instances = mock(InstancesMetadata.class);
+ when(instances.instances()).thenReturn(Collections.singletonList(instanceMetadata));
+ when(instances.instanceFromId(IDENTIFIER)).thenReturn(instanceMetadata);
+ when(instances.instanceFromHost(LOCALHOST)).thenReturn(instanceMetadata);
+ return instances;
+ }
+
+ @Provides
+ @Singleton
+ @NotNull
+ public EmitterFactory emitterFactory()
+ {
+ return () -> emitter;
+ }
+ }
+
+ private final Injector injector = Guice.createInjector(Modules.override(new MainModule()).with(
+ Modules.override(new TestModule()).with(
+ new ReportSchemaHandlerTestModule())));
+ private WebClient client;
+ private Server server;
+ private JsonEmitter emitter;
+
+ @BeforeEach
+ void before() throws InterruptedException
+ {
+ client = WebClient.create(injector.getInstance(Vertx.class));
+ server = injector.getInstance(Server.class);
+
+ VertxTestContext context = new VertxTestContext();
+ server.start()
+ .onSuccess(result -> context.completeNow())
+ .onFailure(context::failNow);
+ context.awaitCompletion(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @AfterEach
+ @SuppressWarnings("ResultOfMethodCallIgnored")
+ @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED")
Review Comment:
Removed the first one, the second one has to stay as it's treated as `ERROR` by SpotBugs otherwise.
##########
server/src/test/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandlerTest.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import com.datastax.driver.core.Metadata;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Provides;
+import com.google.inject.Singleton;
+import com.google.inject.util.Modules;
+import com.linkedin.mxe.MetadataChangeProposal;
+import datahub.client.Callback;
+import datahub.client.MetadataWriteResponse;
+import datahub.shaded.findbugs.annotations.SuppressFBWarnings;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.Vertx;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.client.predicate.ResponsePredicate;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+import org.apache.cassandra.sidecar.TestModule;
+import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
+import org.apache.cassandra.sidecar.cluster.InstancesMetadata;
+import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.utils.IOUtils;
+import org.apache.cassandra.sidecar.datahub.EmitterFactory;
+import org.apache.cassandra.sidecar.datahub.JsonEmitter;
+import org.apache.cassandra.sidecar.server.MainModule;
+import org.apache.cassandra.sidecar.server.Server;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link ReportSchemaHandler}
+ */
+@ExtendWith(VertxExtension.class)
+final class ReportSchemaHandlerTest
+{
+ private static final String CLUSTER = "cluster";
+ private static final String DIRECTORY = "/tmp";
+ private static final int IDENTIFIER = 42;
+ private static final String LOCALHOST = "127.0.0.1";
+ private static final int PORT = 9042;
+ private static final String ENDPOINT = "/api/v1/report-schema";
+ private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+ private static final class ThrowingEmitter extends JsonEmitter
+ {
+ @Override
+ @NotNull
+ public synchronized Future<MetadataWriteResponse> emit(@NotNull MetadataChangeProposal proposal,
+ @Nullable Callback callback) throws IOException
+ {
+ throw new IOException();
+ }
+ }
+
+ private final class ReportSchemaHandlerTestModule extends AbstractModule
+ {
+ @Provides
+ @Singleton
+ @NotNull
+ public InstancesMetadata instancesMetadata()
+ {
+ Metadata metadata = mock(Metadata.class);
+ when(metadata.getKeyspaces()).thenReturn(Collections.emptyList());
+
+ StorageOperations operations = mock(StorageOperations.class);
+ when(operations.clusterName()).thenReturn(CLUSTER);
+
+ CassandraAdapterDelegate delegate = mock(CassandraAdapterDelegate.class);
+ when(delegate.storageOperations()).thenReturn(operations);
+ when(delegate.metadata()).thenReturn(metadata);
+
+ InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
+ when(instanceMetadata.stagingDir()).thenReturn(DIRECTORY);
+ when(instanceMetadata.id()).thenReturn(IDENTIFIER);
+ when(instanceMetadata.host()).thenReturn(LOCALHOST);
+ when(instanceMetadata.port()).thenReturn(PORT);
+ when(instanceMetadata.delegate()).thenReturn(delegate);
+
+ InstancesMetadata instances = mock(InstancesMetadata.class);
+ when(instances.instances()).thenReturn(Collections.singletonList(instanceMetadata));
+ when(instances.instanceFromId(IDENTIFIER)).thenReturn(instanceMetadata);
+ when(instances.instanceFromHost(LOCALHOST)).thenReturn(instanceMetadata);
+ return instances;
+ }
+
+ @Provides
+ @Singleton
+ @NotNull
+ public EmitterFactory emitterFactory()
+ {
+ return () -> emitter;
+ }
+ }
+
+ private final Injector injector = Guice.createInjector(Modules.override(new MainModule()).with(
+ Modules.override(new TestModule()).with(
+ new ReportSchemaHandlerTestModule())));
+ private WebClient client;
+ private Server server;
+ private JsonEmitter emitter;
+
+ @BeforeEach
+ void before() throws InterruptedException
+ {
+ client = WebClient.create(injector.getInstance(Vertx.class));
+ server = injector.getInstance(Server.class);
+
+ VertxTestContext context = new VertxTestContext();
+ server.start()
+ .onSuccess(result -> context.completeNow())
+ .onFailure(context::failNow);
+ context.awaitCompletion(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @AfterEach
+ @SuppressWarnings("ResultOfMethodCallIgnored")
+ @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED")
+ void after() throws InterruptedException
+ {
+ CountDownLatch latch = new CountDownLatch(1);
+ server.close()
+ .onSuccess(future -> latch.countDown());
+ latch.await(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @Test
+ @SuppressWarnings("deprecation")
Review Comment:
Removed.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1987923785
##########
CHANGES.txt:
##########
Review Comment:
Rebased.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1988209774
##########
server/src/main/java/org/apache/cassandra/sidecar/datahub/SchemaReporter.java:
##########
@@ -113,13 +113,23 @@ protected SchemaReporter(@NotNull IdentifiersProvider identifiersProvider,
/**
* Public method for converting and reporting the Cassandra schema
*
- * @param cluster a {@link Cluster} to extract Cassandra schema from
+ * @param cluster the {@link Cluster} to extract Cassandra schema from
*/
public void process(@NotNull Cluster cluster)
+ {
+ process(cluster.getMetadata());
+ }
+
+ /**
+ * Public method for converting and reporting the Cassandra schema
+ *
+ * @param metadata the {@link Metadata} to extract Cassandra schema from
+ */
+ public void process(@NotNull Metadata metadata)
{
try (Emitter emitter = emitterFactory.emitter())
{
- stream(cluster.getMetadata())
+ stream(metadata)
.forEach(ThrowableUtils.consumer(emitter::emit));
Review Comment:
It's way easier to read *and* put breakpoints in this way.
Giving it another thought, three lines make even more sense, updated.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "frankgh (via GitHub)" <gi...@apache.org>.
frankgh commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1963977957
##########
server/src/main/java/org/apache/cassandra/sidecar/acl/authorization/BasicPermissions.java:
##########
@@ -63,6 +63,9 @@ public class BasicPermissions
public static final Permission READ_OPERATIONAL_JOB = new DomainAwarePermission("OPERATIONAL_JOB:READ", OPERATION_SCOPE);
public static final Permission DECOMMISSION_NODE = new DomainAwarePermission("NODE:DECOMMISSION", OPERATION_SCOPE);
+ // Permissions related to Schema Reporting
+ public static final Permission REPORT_SCHEMA = new DomainAwarePermission("SCHEMA:REPORT", CLUSTER_SCOPE);
Review Comment:
The permission is not for someone to read the schema, but for someone to trigger the schema report on demand. So I think the permission is different. Would love to hear input from @sarankk
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1987779473
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/ApiEndpointsV1.java:
##########
@@ -121,13 +121,16 @@ public final class ApiEndpointsV1
public static final String LIST_CDC_SEGMENTS_ROUTE = API_V1 + CDC_PATH + "/segments";
public static final String STREAM_CDC_SEGMENTS_ROUTE = LIST_CDC_SEGMENTS_ROUTE + "/" + SEGMENT_PATH_PARAM;
+ // Schema Reporting
+ private static final String REPORT_SCHEMA = "/report-schema";
+ public static final String REPORT_SCHEMA_ROUTE = API_V1 + REPORT_SCHEMA;
Review Comment:
The `schema_reporting` in sidecar.yml is to configure the scheduled task, which is different from the API.
IMO, `PUT .../schemas` means publishing to the `schemas` resource already. The `publishing` portion is redundant. Having the same term in the configuration and API is not a goal and might be impractical to achieve, since the API scheme and ymal are unrelated.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1987912895
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandler.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.util.Collections;
+import java.util.Set;
+
+import com.datastax.driver.core.Metadata;
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.auth.authorization.Authorization;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.acl.authorization.BasicPermissions;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.datahub.SchemaReporter;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * An implementation of {@link AbstractHandler} used to trigger an immediate,
+ * synchronous conversion and report of the current schema
+ */
+@Singleton
+public class ReportSchemaHandler extends AbstractHandler<Void> implements AccessProtected
+{
+ @NotNull
+ private final SchemaReporter schemaReporter;
+
+ /**
+ * Constructs a new instance of {@link ReportSchemaHandler} using the provided instances
+ * of {@link InstanceMetadataFetcher}, {@link ExecutorPools}, and {@link SchemaReporter}
+ *
+ * @param metadata the metadata fetcher
+ * @param executor executor pools for blocking executions
+ * @param reporter schema reporter to use for the conversion
+ */
+ @Inject
+ public ReportSchemaHandler(@NotNull InstanceMetadataFetcher metadata,
+ @NotNull ExecutorPools executor,
+ @NotNull SchemaReporter reporter)
+ {
+ super(metadata, executor, null);
+
+ schemaReporter = reporter;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ @NotNull
+ public Set<Authorization> requiredAuthorizations()
+ {
+ return Collections.singleton(BasicPermissions.REPORT_SCHEMA.toAuthorization());
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ @Nullable
+ protected Void extractParamsOrThrow(@NotNull RoutingContext context)
+ {
+ return null;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected void handleInternal(@NotNull RoutingContext context,
+ @NotNull HttpServerRequest http,
+ @NotNull String host,
+ @NotNull SocketAddress address,
+ @Nullable Void request)
+ {
+ Metadata metadata = metadataFetcher.callOnFirstAvailableInstance(instance -> instance.delegate().metadata());
+
+ executorPools.service()
+ .runBlocking(() -> schemaReporter.process(metadata))
+ .onSuccess(ignored -> context.end())
Review Comment:
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1984017024
##########
server/src/test/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandlerTest.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import com.datastax.driver.core.Metadata;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Provides;
+import com.google.inject.Singleton;
+import com.google.inject.util.Modules;
+import com.linkedin.mxe.MetadataChangeProposal;
+import datahub.client.Callback;
+import datahub.client.MetadataWriteResponse;
+import datahub.shaded.findbugs.annotations.SuppressFBWarnings;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.Vertx;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.client.predicate.ResponsePredicate;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+import org.apache.cassandra.sidecar.TestModule;
+import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
+import org.apache.cassandra.sidecar.cluster.InstancesMetadata;
+import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.utils.IOUtils;
+import org.apache.cassandra.sidecar.datahub.EmitterFactory;
+import org.apache.cassandra.sidecar.datahub.JsonEmitter;
+import org.apache.cassandra.sidecar.server.MainModule;
+import org.apache.cassandra.sidecar.server.Server;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link ReportSchemaHandler}
+ */
+@ExtendWith(VertxExtension.class)
+final class ReportSchemaHandlerTest
+{
+ private static final String CLUSTER = "cluster";
+ private static final String DIRECTORY = "/tmp";
+ private static final int IDENTIFIER = 42;
+ private static final String LOCALHOST = "127.0.0.1";
+ private static final int PORT = 9042;
+ private static final String ENDPOINT = "/api/v1/report-schema";
+ private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+ private static final class ThrowingEmitter extends JsonEmitter
+ {
+ @Override
+ @NotNull
+ public synchronized Future<MetadataWriteResponse> emit(@NotNull MetadataChangeProposal proposal,
+ @Nullable Callback callback) throws IOException
+ {
+ throw new IOException();
+ }
+ }
+
+ private final class ReportSchemaHandlerTestModule extends AbstractModule
+ {
+ @Provides
+ @Singleton
+ @NotNull
+ public InstancesMetadata instancesMetadata()
+ {
+ Metadata metadata = mock(Metadata.class);
+ when(metadata.getKeyspaces()).thenReturn(Collections.emptyList());
+
+ StorageOperations operations = mock(StorageOperations.class);
+ when(operations.clusterName()).thenReturn(CLUSTER);
+
+ CassandraAdapterDelegate delegate = mock(CassandraAdapterDelegate.class);
+ when(delegate.storageOperations()).thenReturn(operations);
+ when(delegate.metadata()).thenReturn(metadata);
+
+ InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
+ when(instanceMetadata.stagingDir()).thenReturn(DIRECTORY);
+ when(instanceMetadata.id()).thenReturn(IDENTIFIER);
+ when(instanceMetadata.host()).thenReturn(LOCALHOST);
+ when(instanceMetadata.port()).thenReturn(PORT);
+ when(instanceMetadata.delegate()).thenReturn(delegate);
+
+ InstancesMetadata instances = mock(InstancesMetadata.class);
+ when(instances.instances()).thenReturn(Collections.singletonList(instanceMetadata));
+ when(instances.instanceFromId(IDENTIFIER)).thenReturn(instanceMetadata);
+ when(instances.instanceFromHost(LOCALHOST)).thenReturn(instanceMetadata);
+ return instances;
+ }
+
+ @Provides
+ @Singleton
+ @NotNull
+ public EmitterFactory emitterFactory()
+ {
+ return () -> emitter;
+ }
+ }
+
+ private final Injector injector = Guice.createInjector(Modules.override(new MainModule()).with(
+ Modules.override(new TestModule()).with(
+ new ReportSchemaHandlerTestModule())));
+ private WebClient client;
+ private Server server;
+ private JsonEmitter emitter;
+
+ @BeforeEach
+ void before() throws InterruptedException
+ {
+ client = WebClient.create(injector.getInstance(Vertx.class));
+ server = injector.getInstance(Server.class);
+
+ VertxTestContext context = new VertxTestContext();
+ server.start()
+ .onSuccess(result -> context.completeNow())
+ .onFailure(context::failNow);
+ context.awaitCompletion(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @AfterEach
+ @SuppressWarnings("ResultOfMethodCallIgnored")
+ @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED")
+ void after() throws InterruptedException
+ {
+ CountDownLatch latch = new CountDownLatch(1);
+ server.close()
+ .onSuccess(future -> latch.countDown());
+ latch.await(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @Test
+ @SuppressWarnings("deprecation")
+ void testSuccess(@NotNull VertxTestContext context) throws IOException
+ {
+ String expected = IOUtils.readFully("/datahub/empty_cluster.json");
+ emitter = new JsonEmitter();
+
+ client.get(server.actualPort(), LOCALHOST, ENDPOINT)
+ .expect(ResponsePredicate.SC_OK)
+ .send(context.succeeding(response ->
+ {
+ assertThat(response.statusCode())
+ .isEqualTo(HttpResponseStatus.OK.code());
+ assertThat(emitter.content())
+ .isEqualTo(expected);
+ context.completeNow();
+ }));
+ }
+
+ @Test
+ @SuppressWarnings("deprecation")
Review Comment:
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1984014091
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandler.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.util.Collections;
+import java.util.Set;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.auth.authorization.Authorization;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.acl.authorization.BasicPermissions;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.datahub.SchemaReporter;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * An implementation of {@link AbstractHandler} used to trigger an immediate,
+ * synchronous conversion and report of the current schema
+ */
+@Singleton
+public class ReportSchemaHandler extends AbstractHandler<Void> implements AccessProtected
+{
+ @NotNull
+ private final SchemaReporter schemaReporter;
+
+ /**
+ * Constructs a new instance of {@link ReportSchemaHandler} using the provided instances
+ * of {@link InstanceMetadataFetcher}, {@link ExecutorPools}, and {@link SchemaReporter}
+ *
+ * @param metadata the metadata fetcher
+ * @param executor executor pools for blocking executions
+ * @param reporter executor pools for blocking executions
+ */
+ @Inject
+ public ReportSchemaHandler(@NotNull InstanceMetadataFetcher metadata,
+ @NotNull ExecutorPools executor,
+ @NotNull SchemaReporter reporter)
+ {
+ super(metadata, executor, null);
+
+ schemaReporter = reporter;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ @NotNull
+ public Set<Authorization> requiredAuthorizations()
+ {
+ return Collections.singleton(BasicPermissions.REPORT_SCHEMA.toAuthorization());
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ @Nullable
+ protected Void extractParamsOrThrow(@NotNull RoutingContext context)
+ {
+ return null;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected void handleInternal(@NotNull RoutingContext context,
+ @NotNull HttpServerRequest http,
+ @NotNull String host,
+ @NotNull SocketAddress address,
+ @Nullable Void request)
+ {
+ executorPools.service()
+ .runBlocking(() -> metadataFetcher.runOnFirstAvailableInstance(instance ->
+ schemaReporter.process(instance.delegate().metadata())))
+ .onSuccess(context::json)
Review Comment:
I changed this to `context.end()`, no difference in behavior.
An empty response with status code 200 is fine here, in my opinion.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1989718863
##########
server/src/main/java/org/apache/cassandra/sidecar/datahub/SchemaReporter.java:
##########
@@ -113,13 +113,23 @@ protected SchemaReporter(@NotNull IdentifiersProvider identifiersProvider,
/**
* Public method for converting and reporting the Cassandra schema
*
- * @param cluster a {@link Cluster} to extract Cassandra schema from
+ * @param cluster the {@link Cluster} to extract Cassandra schema from
*/
public void process(@NotNull Cluster cluster)
+ {
+ process(cluster.getMetadata());
+ }
+
+ /**
+ * Public method for converting and reporting the Cassandra schema
+ *
+ * @param metadata the {@link Metadata} to extract Cassandra schema from
+ */
+ public void process(@NotNull Metadata metadata)
{
try (Emitter emitter = emitterFactory.emitter())
{
- stream(cluster.getMetadata())
+ stream(metadata)
.forEach(ThrowableUtils.consumer(emitter::emit));
Review Comment:
If in 2 lines, the extra indentation at the second line should be removed.
```suggestion
stream(metadata)
.forEach(ThrowableUtils.consumer(emitter::emit));
```
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "frankgh (via GitHub)" <gi...@apache.org>.
frankgh commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1987746653
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/ApiEndpointsV1.java:
##########
@@ -121,13 +121,16 @@ public final class ApiEndpointsV1
public static final String LIST_CDC_SEGMENTS_ROUTE = API_V1 + CDC_PATH + "/segments";
public static final String STREAM_CDC_SEGMENTS_ROUTE = LIST_CDC_SEGMENTS_ROUTE + "/" + SEGMENT_PATH_PARAM;
+ // Schema Reporting
+ private static final String REPORT_SCHEMA = "/report-schema";
+ public static final String REPORT_SCHEMA_ROUTE = API_V1 + REPORT_SCHEMA;
Review Comment:
`schema-reporting` would map directly to our configuration in `sidecar.yaml`, so I'd prefer to have some consistency between what's user facing (sidecar.yaml) and the endpoints . 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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1988208028
##########
server/src/test/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandlerTest.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import com.datastax.driver.core.Metadata;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Provides;
+import com.google.inject.Singleton;
+import com.google.inject.util.Modules;
+import com.linkedin.mxe.MetadataChangeProposal;
+import datahub.client.Callback;
+import datahub.client.MetadataWriteResponse;
+import datahub.shaded.findbugs.annotations.SuppressFBWarnings;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.Vertx;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.client.predicate.ResponsePredicate;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+import org.apache.cassandra.sidecar.TestModule;
+import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
+import org.apache.cassandra.sidecar.cluster.InstancesMetadata;
+import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.utils.IOUtils;
+import org.apache.cassandra.sidecar.datahub.EmitterFactory;
+import org.apache.cassandra.sidecar.datahub.JsonEmitter;
+import org.apache.cassandra.sidecar.server.MainModule;
+import org.apache.cassandra.sidecar.server.Server;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link ReportSchemaHandler}
+ */
+@ExtendWith(VertxExtension.class)
+final class ReportSchemaHandlerTest
+{
+ private static final String CLUSTER = "cluster";
+ private static final String DIRECTORY = "/tmp";
+ private static final int IDENTIFIER = 42;
+ private static final String LOCALHOST = "127.0.0.1";
+ private static final int PORT = 9042;
+ private static final String ENDPOINT = "/api/v1/report-schema";
+ private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+ private static final class ThrowingEmitter extends JsonEmitter
+ {
+ @Override
+ @NotNull
+ public synchronized Future<MetadataWriteResponse> emit(@NotNull MetadataChangeProposal proposal,
+ @Nullable Callback callback) throws IOException
+ {
+ throw new IOException();
+ }
+ }
+
+ private final class ReportSchemaHandlerTestModule extends AbstractModule
+ {
+ @Provides
+ @Singleton
+ @NotNull
+ public InstancesMetadata instancesMetadata()
+ {
+ Metadata metadata = mock(Metadata.class);
+ when(metadata.getKeyspaces()).thenReturn(Collections.emptyList());
+
+ StorageOperations operations = mock(StorageOperations.class);
+ when(operations.clusterName()).thenReturn(CLUSTER);
+
+ CassandraAdapterDelegate delegate = mock(CassandraAdapterDelegate.class);
+ when(delegate.storageOperations()).thenReturn(operations);
+ when(delegate.metadata()).thenReturn(metadata);
+
+ InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
+ when(instanceMetadata.stagingDir()).thenReturn(DIRECTORY);
+ when(instanceMetadata.id()).thenReturn(IDENTIFIER);
+ when(instanceMetadata.host()).thenReturn(LOCALHOST);
+ when(instanceMetadata.port()).thenReturn(PORT);
+ when(instanceMetadata.delegate()).thenReturn(delegate);
+
+ InstancesMetadata instances = mock(InstancesMetadata.class);
+ when(instances.instances()).thenReturn(Collections.singletonList(instanceMetadata));
+ when(instances.instanceFromId(IDENTIFIER)).thenReturn(instanceMetadata);
+ when(instances.instanceFromHost(LOCALHOST)).thenReturn(instanceMetadata);
+ return instances;
+ }
+
+ @Provides
+ @Singleton
+ @NotNull
+ public EmitterFactory emitterFactory()
+ {
+ return () -> emitter;
+ }
+ }
+
+ private final Injector injector = Guice.createInjector(Modules.override(new MainModule()).with(
+ Modules.override(new TestModule()).with(
+ new ReportSchemaHandlerTestModule())));
+ private WebClient client;
+ private Server server;
+ private JsonEmitter emitter;
+
+ @BeforeEach
+ void before() throws InterruptedException
+ {
+ VertxTestContext context = new VertxTestContext();
+ server = injector.getInstance(Server.class);
+ server.start()
+ .onSuccess(result -> context.completeNow())
+ .onFailure(context::failNow);
+
+ client = WebClient.create(injector.getInstance(Vertx.class));
+
+ context.awaitCompletion(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @AfterEach
+ @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED")
+ void after() throws InterruptedException
+ {
+ CountDownLatch latch = new CountDownLatch(1);
+ server.close()
+ .onSuccess(future -> latch.countDown());
+
+ client.close();
+
+ latch.await(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @Test
+ void testSuccess(@NotNull VertxTestContext context) throws IOException
+ {
+ String expected = IOUtils.readFully("/datahub/empty_cluster.json");
+ emitter = new JsonEmitter();
+ assertThat(emitter.content().length())
+ .isLessThanOrEqualTo(1);
+
+ client.put(server.actualPort(), LOCALHOST, ENDPOINT)
+ .expect(ResponsePredicate.SC_OK)
+ .send(context.succeeding(response ->
+ {
+ assertThat(response.statusCode())
+ .isEqualTo(HttpResponseStatus.OK.code());
+ assertThat(emitter.content())
+ .isEqualTo(expected);
+ context.completeNow();
+ }));
Review Comment:
Running `./gradlew idea` doesn't seem to update any files.
Just to confirm, you'd like me to apply automatic formatting to all files touched in this PR?
This will go against your desire to _"keep the patch succinct, i.e. no unrelated changes"_, as the formatting currently seems to be pretty inconsistent…
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1988205004
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/ApiEndpointsV1.java:
##########
@@ -121,13 +121,16 @@ public final class ApiEndpointsV1
public static final String LIST_CDC_SEGMENTS_ROUTE = API_V1 + CDC_PATH + "/segments";
public static final String STREAM_CDC_SEGMENTS_ROUTE = LIST_CDC_SEGMENTS_ROUTE + "/" + SEGMENT_PATH_PARAM;
+ // Schema Reporting
+ private static final String REPORT_SCHEMA = "/report-schema";
+ public static final String REPORT_SCHEMA_ROUTE = API_V1 + REPORT_SCHEMA;
Review Comment:
I get your point, can you specify the exact string you want me put in there, so there's no confusion.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1989997976
##########
server/src/test/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandlerTest.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import com.datastax.driver.core.Metadata;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Provides;
+import com.google.inject.Singleton;
+import com.google.inject.util.Modules;
+import com.linkedin.mxe.MetadataChangeProposal;
+import datahub.client.Callback;
+import datahub.client.MetadataWriteResponse;
+import datahub.shaded.findbugs.annotations.SuppressFBWarnings;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.Vertx;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.client.predicate.ResponsePredicate;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+import org.apache.cassandra.sidecar.TestModule;
+import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
+import org.apache.cassandra.sidecar.cluster.InstancesMetadata;
+import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.utils.IOUtils;
+import org.apache.cassandra.sidecar.datahub.EmitterFactory;
+import org.apache.cassandra.sidecar.datahub.JsonEmitter;
+import org.apache.cassandra.sidecar.server.MainModule;
+import org.apache.cassandra.sidecar.server.Server;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link ReportSchemaHandler}
+ */
+@ExtendWith(VertxExtension.class)
+final class ReportSchemaHandlerTest
+{
+ private static final String CLUSTER = "cluster";
+ private static final String DIRECTORY = "/tmp";
+ private static final int IDENTIFIER = 42;
+ private static final String LOCALHOST = "127.0.0.1";
+ private static final int PORT = 9042;
+ private static final String ENDPOINT = "/api/v1/report-schema";
+ private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+ private static final class ThrowingEmitter extends JsonEmitter
+ {
+ @Override
+ @NotNull
+ public synchronized Future<MetadataWriteResponse> emit(@NotNull MetadataChangeProposal proposal,
+ @Nullable Callback callback) throws IOException
+ {
+ throw new IOException();
+ }
+ }
+
+ private final class ReportSchemaHandlerTestModule extends AbstractModule
+ {
+ @Provides
+ @Singleton
+ @NotNull
+ public InstancesMetadata instancesMetadata()
+ {
+ Metadata metadata = mock(Metadata.class);
+ when(metadata.getKeyspaces()).thenReturn(Collections.emptyList());
+
+ StorageOperations operations = mock(StorageOperations.class);
+ when(operations.clusterName()).thenReturn(CLUSTER);
+
+ CassandraAdapterDelegate delegate = mock(CassandraAdapterDelegate.class);
+ when(delegate.storageOperations()).thenReturn(operations);
+ when(delegate.metadata()).thenReturn(metadata);
+
+ InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
+ when(instanceMetadata.stagingDir()).thenReturn(DIRECTORY);
+ when(instanceMetadata.id()).thenReturn(IDENTIFIER);
+ when(instanceMetadata.host()).thenReturn(LOCALHOST);
+ when(instanceMetadata.port()).thenReturn(PORT);
+ when(instanceMetadata.delegate()).thenReturn(delegate);
+
+ InstancesMetadata instances = mock(InstancesMetadata.class);
+ when(instances.instances()).thenReturn(Collections.singletonList(instanceMetadata));
+ when(instances.instanceFromId(IDENTIFIER)).thenReturn(instanceMetadata);
+ when(instances.instanceFromHost(LOCALHOST)).thenReturn(instanceMetadata);
+ return instances;
+ }
+
+ @Provides
+ @Singleton
+ @NotNull
+ public EmitterFactory emitterFactory()
+ {
+ return () -> emitter;
+ }
+ }
+
+ private final Injector injector = Guice.createInjector(Modules.override(new MainModule()).with(
+ Modules.override(new TestModule()).with(
+ new ReportSchemaHandlerTestModule())));
+ private WebClient client;
+ private Server server;
+ private JsonEmitter emitter;
+
+ @BeforeEach
+ void before() throws InterruptedException
+ {
+ VertxTestContext context = new VertxTestContext();
+ server = injector.getInstance(Server.class);
+ server.start()
+ .onSuccess(result -> context.completeNow())
+ .onFailure(context::failNow);
+
+ client = WebClient.create(injector.getInstance(Vertx.class));
+
+ context.awaitCompletion(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @AfterEach
+ @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED")
+ void after() throws InterruptedException
+ {
+ CountDownLatch latch = new CountDownLatch(1);
+ server.close()
+ .onSuccess(future -> latch.countDown());
+
+ client.close();
+
+ latch.await(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @Test
+ void testSuccess(@NotNull VertxTestContext context) throws IOException
+ {
+ String expected = IOUtils.readFully("/datahub/empty_cluster.json");
+ emitter = new JsonEmitter();
+ assertThat(emitter.content().length())
+ .isLessThanOrEqualTo(1);
+
+ client.put(server.actualPort(), LOCALHOST, ENDPOINT)
+ .expect(ResponsePredicate.SC_OK)
+ .send(context.succeeding(response ->
+ {
+ assertThat(response.statusCode())
+ .isEqualTo(HttpResponseStatus.OK.code());
+ assertThat(emitter.content())
+ .isEqualTo(expected);
+ context.completeNow();
+ }));
Review Comment:
Applied the required identations to all files affected by 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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1962766807
##########
CHANGES.txt:
##########
@@ -8,7 +9,7 @@
* Sidecar schema initialization can be executed on multiple thread (CASSSIDECAR-200)
* Make sidecar operations resilient to down Cassandra nodes (CASSSIDECAR-201)
* Fix Cassandra instance not found error (CASSSIDECAR-192)
- * Implemented Schema Reporter for Integration with DataHub (CASSSIDECAR-191)
+ * Implement Schema Reporter for Integration with DataHub (CASSSIDECAR-191)
Review Comment:
Do not change the existing entries in the `CHANGES.txt`
##########
server/src/main/java/org/apache/cassandra/sidecar/utils/InstanceMetadataFetcher.java:
##########
@@ -97,13 +97,29 @@ public CassandraAdapterDelegate delegate(@NotNull String host) throws NoSuchCass
}
/**
- * Iterate through the local instances and call the function on the first available instance, i.e. no CassandraUnavailableException
- * or OperationUnavailableException is thrown for the operations
+ * Iterate through the local instances and run the {@link Consumer} on the first available one,
+ * so no {@link CassandraUnavailableException} or {@link OperationUnavailableException} is thrown for the operations
+ *
+ * @param consumer a {@link Consumer} that processes {@link InstanceMetadata} and returns no result
+ * @throws CassandraUnavailableException if all local instances were exhausted
+ */
+ public void runOnFirstAvailableInstance(Consumer<InstanceMetadata> consumer) throws CassandraUnavailableException
+ {
+ callOnFirstAvailableInstance(metadata ->
+ {
+ consumer.accept(metadata);
+ return null;
+ });
+ }
Review Comment:
Can you remove this method? It is unnecessary. It does not retrieve anything from instance metadata fetcher, who is supposed to "fetch something". I have suggested a different implementation in the new handler w/o using this method.
##########
server/src/main/java/org/apache/cassandra/sidecar/acl/authorization/BasicPermissions.java:
##########
@@ -63,6 +63,9 @@ public class BasicPermissions
public static final Permission READ_OPERATIONAL_JOB = new DomainAwarePermission("OPERATIONAL_JOB:READ", OPERATION_SCOPE);
public static final Permission DECOMMISSION_NODE = new DomainAwarePermission("NODE:DECOMMISSION", OPERATION_SCOPE);
+ // Permissions related to Schema Reporting
+ public static final Permission REPORT_SCHEMA = new DomainAwarePermission("SCHEMA:REPORT", CLUSTER_SCOPE);
Review Comment:
I do not think an extra permission is required.
The permission needed in order to publish to DataHub is `SCHEMA:READ`, which already exists. cc: @sarankk
##########
server/src/main/java/org/apache/cassandra/sidecar/server/MainModule.java:
##########
@@ -902,7 +912,8 @@ public IdentifiersProvider identifiersProvider(@NotNull InstanceMetadataFetcher
@NotNull
protected String initialize()
{
- return fetcher.callOnFirstAvailableInstance(i -> i.delegate().storageOperations().clusterName());
+ return fetcher.callOnFirstAvailableInstance(instance ->
+ instance.delegate().storageOperations().clusterName());
Review Comment:
Please revert the unrelated change.
##########
server/src/test/java/org/apache/cassandra/sidecar/utils/InstanceMetadataFetcherTest.java:
##########
@@ -21,7 +21,7 @@
import java.nio.file.Path;
import java.util.Arrays;
import java.util.List;
-
+import com.google.common.collect.ImmutableList;
Review Comment:
Revert the changes in this file once you remove `runOnFirstAvailableInstance`
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandler.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.util.Collections;
+import java.util.Set;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.auth.authorization.Authorization;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.acl.authorization.BasicPermissions;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.datahub.SchemaReporter;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * An implementation of {@link AbstractHandler} used to trigger an immediate,
+ * synchronous conversion and report of the current schema
+ */
+@Singleton
+public class ReportSchemaHandler extends AbstractHandler<Void> implements AccessProtected
+{
+ @NotNull
+ private final SchemaReporter schemaReporter;
+
+ /**
+ * Constructs a new instance of {@link ReportSchemaHandler} using the provided instances
+ * of {@link InstanceMetadataFetcher}, {@link ExecutorPools}, and {@link SchemaReporter}
+ *
+ * @param metadata the metadata fetcher
+ * @param executor executor pools for blocking executions
+ * @param reporter executor pools for blocking executions
Review Comment:
copy-paste error
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandler.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.util.Collections;
+import java.util.Set;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.auth.authorization.Authorization;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.acl.authorization.BasicPermissions;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.datahub.SchemaReporter;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * An implementation of {@link AbstractHandler} used to trigger an immediate,
+ * synchronous conversion and report of the current schema
+ */
+@Singleton
+public class ReportSchemaHandler extends AbstractHandler<Void> implements AccessProtected
+{
+ @NotNull
+ private final SchemaReporter schemaReporter;
+
+ /**
+ * Constructs a new instance of {@link ReportSchemaHandler} using the provided instances
+ * of {@link InstanceMetadataFetcher}, {@link ExecutorPools}, and {@link SchemaReporter}
+ *
+ * @param metadata the metadata fetcher
+ * @param executor executor pools for blocking executions
+ * @param reporter executor pools for blocking executions
+ */
+ @Inject
+ public ReportSchemaHandler(@NotNull InstanceMetadataFetcher metadata,
+ @NotNull ExecutorPools executor,
+ @NotNull SchemaReporter reporter)
+ {
+ super(metadata, executor, null);
+
+ schemaReporter = reporter;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ @NotNull
+ public Set<Authorization> requiredAuthorizations()
+ {
+ return Collections.singleton(BasicPermissions.REPORT_SCHEMA.toAuthorization());
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ @Nullable
+ protected Void extractParamsOrThrow(@NotNull RoutingContext context)
+ {
+ return null;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected void handleInternal(@NotNull RoutingContext context,
+ @NotNull HttpServerRequest http,
+ @NotNull String host,
+ @NotNull SocketAddress address,
+ @Nullable Void request)
+ {
+ executorPools.service()
+ .runBlocking(() -> metadataFetcher.runOnFirstAvailableInstance(instance ->
+ schemaReporter.process(instance.delegate().metadata())))
+ .onSuccess(context::json)
Review Comment:
The previous block does not return any value. What does the response json look like?
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandler.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.util.Collections;
+import java.util.Set;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.auth.authorization.Authorization;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.acl.authorization.BasicPermissions;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.datahub.SchemaReporter;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * An implementation of {@link AbstractHandler} used to trigger an immediate,
+ * synchronous conversion and report of the current schema
+ */
+@Singleton
+public class ReportSchemaHandler extends AbstractHandler<Void> implements AccessProtected
+{
+ @NotNull
+ private final SchemaReporter schemaReporter;
+
+ /**
+ * Constructs a new instance of {@link ReportSchemaHandler} using the provided instances
+ * of {@link InstanceMetadataFetcher}, {@link ExecutorPools}, and {@link SchemaReporter}
+ *
+ * @param metadata the metadata fetcher
+ * @param executor executor pools for blocking executions
+ * @param reporter executor pools for blocking executions
+ */
+ @Inject
+ public ReportSchemaHandler(@NotNull InstanceMetadataFetcher metadata,
+ @NotNull ExecutorPools executor,
+ @NotNull SchemaReporter reporter)
+ {
+ super(metadata, executor, null);
+
+ schemaReporter = reporter;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ @NotNull
+ public Set<Authorization> requiredAuthorizations()
+ {
+ return Collections.singleton(BasicPermissions.REPORT_SCHEMA.toAuthorization());
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ @Nullable
+ protected Void extractParamsOrThrow(@NotNull RoutingContext context)
+ {
+ return null;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected void handleInternal(@NotNull RoutingContext context,
+ @NotNull HttpServerRequest http,
+ @NotNull String host,
+ @NotNull SocketAddress address,
+ @Nullable Void request)
+ {
+ executorPools.service()
+ .runBlocking(() -> metadataFetcher.runOnFirstAvailableInstance(instance ->
+ schemaReporter.process(instance.delegate().metadata())))
+ .onSuccess(context::json)
+ .onFailure(throwable -> processFailure(throwable, context, host, address, request));
+ }
Review Comment:
```suggestion
Metadata metadata = metadataFetcher.callOnFirstAvailableInstance(instance -> instance.delegate().metadata());
executorPools.service()
.runBlocking(() -> {
schemaReporter.process(metadata);
})
.onSuccess(v -> context.json(OK_STATUS))
.onFailure(cause -> processFailure(cause, context, host, address, request));
```
##########
server/src/test/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandlerTest.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import com.datastax.driver.core.Metadata;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Provides;
+import com.google.inject.Singleton;
+import com.google.inject.util.Modules;
+import com.linkedin.mxe.MetadataChangeProposal;
+import datahub.client.Callback;
+import datahub.client.MetadataWriteResponse;
+import datahub.shaded.findbugs.annotations.SuppressFBWarnings;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.Vertx;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.client.predicate.ResponsePredicate;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+import org.apache.cassandra.sidecar.TestModule;
+import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
+import org.apache.cassandra.sidecar.cluster.InstancesMetadata;
+import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.utils.IOUtils;
+import org.apache.cassandra.sidecar.datahub.EmitterFactory;
+import org.apache.cassandra.sidecar.datahub.JsonEmitter;
+import org.apache.cassandra.sidecar.server.MainModule;
+import org.apache.cassandra.sidecar.server.Server;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link ReportSchemaHandler}
+ */
+@ExtendWith(VertxExtension.class)
+final class ReportSchemaHandlerTest
+{
+ private static final String CLUSTER = "cluster";
+ private static final String DIRECTORY = "/tmp";
+ private static final int IDENTIFIER = 42;
+ private static final String LOCALHOST = "127.0.0.1";
+ private static final int PORT = 9042;
+ private static final String ENDPOINT = "/api/v1/report-schema";
+ private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+ private static final class ThrowingEmitter extends JsonEmitter
+ {
+ @Override
+ @NotNull
+ public synchronized Future<MetadataWriteResponse> emit(@NotNull MetadataChangeProposal proposal,
+ @Nullable Callback callback) throws IOException
+ {
+ throw new IOException();
+ }
+ }
+
+ private final class ReportSchemaHandlerTestModule extends AbstractModule
+ {
+ @Provides
+ @Singleton
+ @NotNull
+ public InstancesMetadata instancesMetadata()
+ {
+ Metadata metadata = mock(Metadata.class);
+ when(metadata.getKeyspaces()).thenReturn(Collections.emptyList());
+
+ StorageOperations operations = mock(StorageOperations.class);
+ when(operations.clusterName()).thenReturn(CLUSTER);
+
+ CassandraAdapterDelegate delegate = mock(CassandraAdapterDelegate.class);
+ when(delegate.storageOperations()).thenReturn(operations);
+ when(delegate.metadata()).thenReturn(metadata);
+
+ InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
+ when(instanceMetadata.stagingDir()).thenReturn(DIRECTORY);
+ when(instanceMetadata.id()).thenReturn(IDENTIFIER);
+ when(instanceMetadata.host()).thenReturn(LOCALHOST);
+ when(instanceMetadata.port()).thenReturn(PORT);
+ when(instanceMetadata.delegate()).thenReturn(delegate);
+
+ InstancesMetadata instances = mock(InstancesMetadata.class);
+ when(instances.instances()).thenReturn(Collections.singletonList(instanceMetadata));
+ when(instances.instanceFromId(IDENTIFIER)).thenReturn(instanceMetadata);
+ when(instances.instanceFromHost(LOCALHOST)).thenReturn(instanceMetadata);
+ return instances;
+ }
+
+ @Provides
+ @Singleton
+ @NotNull
+ public EmitterFactory emitterFactory()
+ {
+ return () -> emitter;
+ }
+ }
+
+ private final Injector injector = Guice.createInjector(Modules.override(new MainModule()).with(
+ Modules.override(new TestModule()).with(
+ new ReportSchemaHandlerTestModule())));
+ private WebClient client;
+ private Server server;
+ private JsonEmitter emitter;
+
+ @BeforeEach
+ void before() throws InterruptedException
+ {
+ client = WebClient.create(injector.getInstance(Vertx.class));
+ server = injector.getInstance(Server.class);
+
+ VertxTestContext context = new VertxTestContext();
+ server.start()
+ .onSuccess(result -> context.completeNow())
+ .onFailure(context::failNow);
+ context.awaitCompletion(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @AfterEach
+ @SuppressWarnings("ResultOfMethodCallIgnored")
+ @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED")
+ void after() throws InterruptedException
+ {
+ CountDownLatch latch = new CountDownLatch(1);
+ server.close()
+ .onSuccess(future -> latch.countDown());
+ latch.await(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @Test
+ @SuppressWarnings("deprecation")
+ void testSuccess(@NotNull VertxTestContext context) throws IOException
+ {
+ String expected = IOUtils.readFully("/datahub/empty_cluster.json");
+ emitter = new JsonEmitter();
+
+ client.get(server.actualPort(), LOCALHOST, ENDPOINT)
+ .expect(ResponsePredicate.SC_OK)
+ .send(context.succeeding(response ->
+ {
+ assertThat(response.statusCode())
+ .isEqualTo(HttpResponseStatus.OK.code());
+ assertThat(emitter.content())
+ .isEqualTo(expected);
+ context.completeNow();
+ }));
+ }
+
+ @Test
+ @SuppressWarnings("deprecation")
Review Comment:
remove
##########
server/src/test/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandlerTest.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import com.datastax.driver.core.Metadata;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Provides;
+import com.google.inject.Singleton;
+import com.google.inject.util.Modules;
+import com.linkedin.mxe.MetadataChangeProposal;
+import datahub.client.Callback;
+import datahub.client.MetadataWriteResponse;
+import datahub.shaded.findbugs.annotations.SuppressFBWarnings;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.Vertx;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.client.predicate.ResponsePredicate;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+import org.apache.cassandra.sidecar.TestModule;
+import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
+import org.apache.cassandra.sidecar.cluster.InstancesMetadata;
+import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.utils.IOUtils;
+import org.apache.cassandra.sidecar.datahub.EmitterFactory;
+import org.apache.cassandra.sidecar.datahub.JsonEmitter;
+import org.apache.cassandra.sidecar.server.MainModule;
+import org.apache.cassandra.sidecar.server.Server;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link ReportSchemaHandler}
+ */
+@ExtendWith(VertxExtension.class)
+final class ReportSchemaHandlerTest
+{
+ private static final String CLUSTER = "cluster";
+ private static final String DIRECTORY = "/tmp";
+ private static final int IDENTIFIER = 42;
+ private static final String LOCALHOST = "127.0.0.1";
+ private static final int PORT = 9042;
+ private static final String ENDPOINT = "/api/v1/report-schema";
+ private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+ private static final class ThrowingEmitter extends JsonEmitter
+ {
+ @Override
+ @NotNull
+ public synchronized Future<MetadataWriteResponse> emit(@NotNull MetadataChangeProposal proposal,
+ @Nullable Callback callback) throws IOException
+ {
+ throw new IOException();
+ }
+ }
+
+ private final class ReportSchemaHandlerTestModule extends AbstractModule
+ {
+ @Provides
+ @Singleton
+ @NotNull
+ public InstancesMetadata instancesMetadata()
+ {
+ Metadata metadata = mock(Metadata.class);
+ when(metadata.getKeyspaces()).thenReturn(Collections.emptyList());
+
+ StorageOperations operations = mock(StorageOperations.class);
+ when(operations.clusterName()).thenReturn(CLUSTER);
+
+ CassandraAdapterDelegate delegate = mock(CassandraAdapterDelegate.class);
+ when(delegate.storageOperations()).thenReturn(operations);
+ when(delegate.metadata()).thenReturn(metadata);
+
+ InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
+ when(instanceMetadata.stagingDir()).thenReturn(DIRECTORY);
+ when(instanceMetadata.id()).thenReturn(IDENTIFIER);
+ when(instanceMetadata.host()).thenReturn(LOCALHOST);
+ when(instanceMetadata.port()).thenReturn(PORT);
+ when(instanceMetadata.delegate()).thenReturn(delegate);
+
+ InstancesMetadata instances = mock(InstancesMetadata.class);
+ when(instances.instances()).thenReturn(Collections.singletonList(instanceMetadata));
+ when(instances.instanceFromId(IDENTIFIER)).thenReturn(instanceMetadata);
+ when(instances.instanceFromHost(LOCALHOST)).thenReturn(instanceMetadata);
+ return instances;
+ }
+
+ @Provides
+ @Singleton
+ @NotNull
+ public EmitterFactory emitterFactory()
+ {
+ return () -> emitter;
+ }
+ }
+
+ private final Injector injector = Guice.createInjector(Modules.override(new MainModule()).with(
+ Modules.override(new TestModule()).with(
+ new ReportSchemaHandlerTestModule())));
+ private WebClient client;
+ private Server server;
+ private JsonEmitter emitter;
+
+ @BeforeEach
+ void before() throws InterruptedException
+ {
+ client = WebClient.create(injector.getInstance(Vertx.class));
+ server = injector.getInstance(Server.class);
+
+ VertxTestContext context = new VertxTestContext();
+ server.start()
+ .onSuccess(result -> context.completeNow())
+ .onFailure(context::failNow);
+ context.awaitCompletion(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @AfterEach
+ @SuppressWarnings("ResultOfMethodCallIgnored")
+ @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED")
+ void after() throws InterruptedException
+ {
+ CountDownLatch latch = new CountDownLatch(1);
+ server.close()
+ .onSuccess(future -> latch.countDown());
+ latch.await(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @Test
+ @SuppressWarnings("deprecation")
Review Comment:
remove
##########
server/src/test/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandlerTest.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import com.datastax.driver.core.Metadata;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Provides;
+import com.google.inject.Singleton;
+import com.google.inject.util.Modules;
+import com.linkedin.mxe.MetadataChangeProposal;
+import datahub.client.Callback;
+import datahub.client.MetadataWriteResponse;
+import datahub.shaded.findbugs.annotations.SuppressFBWarnings;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.Vertx;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.client.predicate.ResponsePredicate;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+import org.apache.cassandra.sidecar.TestModule;
+import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
+import org.apache.cassandra.sidecar.cluster.InstancesMetadata;
+import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.utils.IOUtils;
+import org.apache.cassandra.sidecar.datahub.EmitterFactory;
+import org.apache.cassandra.sidecar.datahub.JsonEmitter;
+import org.apache.cassandra.sidecar.server.MainModule;
+import org.apache.cassandra.sidecar.server.Server;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link ReportSchemaHandler}
+ */
+@ExtendWith(VertxExtension.class)
+final class ReportSchemaHandlerTest
+{
+ private static final String CLUSTER = "cluster";
+ private static final String DIRECTORY = "/tmp";
+ private static final int IDENTIFIER = 42;
+ private static final String LOCALHOST = "127.0.0.1";
+ private static final int PORT = 9042;
+ private static final String ENDPOINT = "/api/v1/report-schema";
+ private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+ private static final class ThrowingEmitter extends JsonEmitter
+ {
+ @Override
+ @NotNull
+ public synchronized Future<MetadataWriteResponse> emit(@NotNull MetadataChangeProposal proposal,
+ @Nullable Callback callback) throws IOException
+ {
+ throw new IOException();
+ }
+ }
+
+ private final class ReportSchemaHandlerTestModule extends AbstractModule
+ {
+ @Provides
+ @Singleton
+ @NotNull
+ public InstancesMetadata instancesMetadata()
+ {
+ Metadata metadata = mock(Metadata.class);
+ when(metadata.getKeyspaces()).thenReturn(Collections.emptyList());
+
+ StorageOperations operations = mock(StorageOperations.class);
+ when(operations.clusterName()).thenReturn(CLUSTER);
+
+ CassandraAdapterDelegate delegate = mock(CassandraAdapterDelegate.class);
+ when(delegate.storageOperations()).thenReturn(operations);
+ when(delegate.metadata()).thenReturn(metadata);
+
+ InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
+ when(instanceMetadata.stagingDir()).thenReturn(DIRECTORY);
+ when(instanceMetadata.id()).thenReturn(IDENTIFIER);
+ when(instanceMetadata.host()).thenReturn(LOCALHOST);
+ when(instanceMetadata.port()).thenReturn(PORT);
+ when(instanceMetadata.delegate()).thenReturn(delegate);
+
+ InstancesMetadata instances = mock(InstancesMetadata.class);
+ when(instances.instances()).thenReturn(Collections.singletonList(instanceMetadata));
+ when(instances.instanceFromId(IDENTIFIER)).thenReturn(instanceMetadata);
+ when(instances.instanceFromHost(LOCALHOST)).thenReturn(instanceMetadata);
+ return instances;
+ }
+
+ @Provides
+ @Singleton
+ @NotNull
+ public EmitterFactory emitterFactory()
+ {
+ return () -> emitter;
+ }
+ }
+
+ private final Injector injector = Guice.createInjector(Modules.override(new MainModule()).with(
+ Modules.override(new TestModule()).with(
+ new ReportSchemaHandlerTestModule())));
+ private WebClient client;
+ private Server server;
+ private JsonEmitter emitter;
+
+ @BeforeEach
+ void before() throws InterruptedException
+ {
+ client = WebClient.create(injector.getInstance(Vertx.class));
+ server = injector.getInstance(Server.class);
+
+ VertxTestContext context = new VertxTestContext();
+ server.start()
+ .onSuccess(result -> context.completeNow())
+ .onFailure(context::failNow);
+ context.awaitCompletion(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @AfterEach
+ @SuppressWarnings("ResultOfMethodCallIgnored")
+ @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED")
Review Comment:
remove the unnecessary annotations.
##########
server/src/test/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandlerTest.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import com.datastax.driver.core.Metadata;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Provides;
+import com.google.inject.Singleton;
+import com.google.inject.util.Modules;
+import com.linkedin.mxe.MetadataChangeProposal;
+import datahub.client.Callback;
+import datahub.client.MetadataWriteResponse;
+import datahub.shaded.findbugs.annotations.SuppressFBWarnings;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.Vertx;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.client.predicate.ResponsePredicate;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+import org.apache.cassandra.sidecar.TestModule;
+import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
+import org.apache.cassandra.sidecar.cluster.InstancesMetadata;
+import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.utils.IOUtils;
+import org.apache.cassandra.sidecar.datahub.EmitterFactory;
+import org.apache.cassandra.sidecar.datahub.JsonEmitter;
+import org.apache.cassandra.sidecar.server.MainModule;
+import org.apache.cassandra.sidecar.server.Server;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link ReportSchemaHandler}
+ */
+@ExtendWith(VertxExtension.class)
+final class ReportSchemaHandlerTest
+{
+ private static final String CLUSTER = "cluster";
+ private static final String DIRECTORY = "/tmp";
+ private static final int IDENTIFIER = 42;
+ private static final String LOCALHOST = "127.0.0.1";
+ private static final int PORT = 9042;
+ private static final String ENDPOINT = "/api/v1/report-schema";
+ private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+ private static final class ThrowingEmitter extends JsonEmitter
+ {
+ @Override
+ @NotNull
+ public synchronized Future<MetadataWriteResponse> emit(@NotNull MetadataChangeProposal proposal,
+ @Nullable Callback callback) throws IOException
+ {
+ throw new IOException();
+ }
+ }
+
+ private final class ReportSchemaHandlerTestModule extends AbstractModule
+ {
+ @Provides
+ @Singleton
+ @NotNull
+ public InstancesMetadata instancesMetadata()
+ {
+ Metadata metadata = mock(Metadata.class);
+ when(metadata.getKeyspaces()).thenReturn(Collections.emptyList());
+
+ StorageOperations operations = mock(StorageOperations.class);
+ when(operations.clusterName()).thenReturn(CLUSTER);
+
+ CassandraAdapterDelegate delegate = mock(CassandraAdapterDelegate.class);
+ when(delegate.storageOperations()).thenReturn(operations);
+ when(delegate.metadata()).thenReturn(metadata);
+
+ InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
+ when(instanceMetadata.stagingDir()).thenReturn(DIRECTORY);
+ when(instanceMetadata.id()).thenReturn(IDENTIFIER);
+ when(instanceMetadata.host()).thenReturn(LOCALHOST);
+ when(instanceMetadata.port()).thenReturn(PORT);
+ when(instanceMetadata.delegate()).thenReturn(delegate);
+
+ InstancesMetadata instances = mock(InstancesMetadata.class);
+ when(instances.instances()).thenReturn(Collections.singletonList(instanceMetadata));
+ when(instances.instanceFromId(IDENTIFIER)).thenReturn(instanceMetadata);
+ when(instances.instanceFromHost(LOCALHOST)).thenReturn(instanceMetadata);
+ return instances;
+ }
+
+ @Provides
+ @Singleton
+ @NotNull
+ public EmitterFactory emitterFactory()
+ {
+ return () -> emitter;
+ }
+ }
+
+ private final Injector injector = Guice.createInjector(Modules.override(new MainModule()).with(
+ Modules.override(new TestModule()).with(
+ new ReportSchemaHandlerTestModule())));
+ private WebClient client;
+ private Server server;
+ private JsonEmitter emitter;
+
+ @BeforeEach
+ void before() throws InterruptedException
+ {
+ client = WebClient.create(injector.getInstance(Vertx.class));
+ server = injector.getInstance(Server.class);
+
+ VertxTestContext context = new VertxTestContext();
+ server.start()
+ .onSuccess(result -> context.completeNow())
+ .onFailure(context::failNow);
+ context.awaitCompletion(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @AfterEach
+ @SuppressWarnings("ResultOfMethodCallIgnored")
+ @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED")
+ void after() throws InterruptedException
+ {
+ CountDownLatch latch = new CountDownLatch(1);
+ server.close()
+ .onSuccess(future -> latch.countDown());
+ latch.await(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
Review Comment:
close the client too.
##########
server/src/test/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandlerTest.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import com.datastax.driver.core.Metadata;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Provides;
+import com.google.inject.Singleton;
+import com.google.inject.util.Modules;
+import com.linkedin.mxe.MetadataChangeProposal;
+import datahub.client.Callback;
+import datahub.client.MetadataWriteResponse;
+import datahub.shaded.findbugs.annotations.SuppressFBWarnings;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.Vertx;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.client.predicate.ResponsePredicate;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+import org.apache.cassandra.sidecar.TestModule;
+import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
+import org.apache.cassandra.sidecar.cluster.InstancesMetadata;
+import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.utils.IOUtils;
+import org.apache.cassandra.sidecar.datahub.EmitterFactory;
+import org.apache.cassandra.sidecar.datahub.JsonEmitter;
+import org.apache.cassandra.sidecar.server.MainModule;
+import org.apache.cassandra.sidecar.server.Server;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link ReportSchemaHandler}
+ */
+@ExtendWith(VertxExtension.class)
+final class ReportSchemaHandlerTest
+{
+ private static final String CLUSTER = "cluster";
+ private static final String DIRECTORY = "/tmp";
+ private static final int IDENTIFIER = 42;
+ private static final String LOCALHOST = "127.0.0.1";
+ private static final int PORT = 9042;
+ private static final String ENDPOINT = "/api/v1/report-schema";
+ private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+ private static final class ThrowingEmitter extends JsonEmitter
+ {
+ @Override
+ @NotNull
+ public synchronized Future<MetadataWriteResponse> emit(@NotNull MetadataChangeProposal proposal,
+ @Nullable Callback callback) throws IOException
+ {
+ throw new IOException();
+ }
+ }
+
+ private final class ReportSchemaHandlerTestModule extends AbstractModule
+ {
+ @Provides
+ @Singleton
+ @NotNull
+ public InstancesMetadata instancesMetadata()
+ {
+ Metadata metadata = mock(Metadata.class);
+ when(metadata.getKeyspaces()).thenReturn(Collections.emptyList());
+
+ StorageOperations operations = mock(StorageOperations.class);
+ when(operations.clusterName()).thenReturn(CLUSTER);
+
+ CassandraAdapterDelegate delegate = mock(CassandraAdapterDelegate.class);
+ when(delegate.storageOperations()).thenReturn(operations);
+ when(delegate.metadata()).thenReturn(metadata);
+
+ InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
+ when(instanceMetadata.stagingDir()).thenReturn(DIRECTORY);
+ when(instanceMetadata.id()).thenReturn(IDENTIFIER);
+ when(instanceMetadata.host()).thenReturn(LOCALHOST);
+ when(instanceMetadata.port()).thenReturn(PORT);
+ when(instanceMetadata.delegate()).thenReturn(delegate);
+
+ InstancesMetadata instances = mock(InstancesMetadata.class);
+ when(instances.instances()).thenReturn(Collections.singletonList(instanceMetadata));
+ when(instances.instanceFromId(IDENTIFIER)).thenReturn(instanceMetadata);
+ when(instances.instanceFromHost(LOCALHOST)).thenReturn(instanceMetadata);
+ return instances;
+ }
+
+ @Provides
+ @Singleton
+ @NotNull
+ public EmitterFactory emitterFactory()
+ {
+ return () -> emitter;
+ }
+ }
+
+ private final Injector injector = Guice.createInjector(Modules.override(new MainModule()).with(
+ Modules.override(new TestModule()).with(
+ new ReportSchemaHandlerTestModule())));
+ private WebClient client;
+ private Server server;
+ private JsonEmitter emitter;
+
+ @BeforeEach
+ void before() throws InterruptedException
+ {
+ client = WebClient.create(injector.getInstance(Vertx.class));
+ server = injector.getInstance(Server.class);
+
+ VertxTestContext context = new VertxTestContext();
+ server.start()
+ .onSuccess(result -> context.completeNow())
+ .onFailure(context::failNow);
+ context.awaitCompletion(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @AfterEach
+ @SuppressWarnings("ResultOfMethodCallIgnored")
+ @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED")
+ void after() throws InterruptedException
+ {
+ CountDownLatch latch = new CountDownLatch(1);
+ server.close()
+ .onSuccess(future -> latch.countDown());
+ latch.await(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @Test
+ @SuppressWarnings("deprecation")
+ void testSuccess(@NotNull VertxTestContext context) throws IOException
+ {
+ String expected = IOUtils.readFully("/datahub/empty_cluster.json");
+ emitter = new JsonEmitter();
+
+ client.get(server.actualPort(), LOCALHOST, ENDPOINT)
Review Comment:
add an assertion that `emitter.content` is empty, before making the http request.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#issuecomment-2670337988
Let's keep the patch succinct, i.e. no unrelated changes and add code only when they are definitely required.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1964179142
##########
CHANGES.txt:
##########
@@ -8,7 +9,7 @@
* Sidecar schema initialization can be executed on multiple thread (CASSSIDECAR-200)
* Make sidecar operations resilient to down Cassandra nodes (CASSSIDECAR-201)
* Fix Cassandra instance not found error (CASSSIDECAR-192)
- * Implemented Schema Reporter for Integration with DataHub (CASSSIDECAR-191)
+ * Implement Schema Reporter for Integration with DataHub (CASSSIDECAR-191)
Review Comment:
Sure.
Is there a specific reason you'd like me to keep this typo I myself accidentally introduced in there forever?
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1988206655
##########
client/src/testFixtures/java/org/apache/cassandra/sidecar/client/SidecarClientTest.java:
##########
@@ -1673,6 +1677,42 @@ public void onError(Throwable throwable)
assertThat(new String(baos.toByteArray(), StandardCharsets.UTF_8)).isEqualTo("Test Content");
}
+ @Test
+ public void testReportSchemaSuccess()
+ {
+ MockResponse response = new MockResponse()
+ .setResponseCode(OK.code())
+ .setBody("");
+
+ enqueue(response);
+
+ SidecarInstance instance = instances.get(0);
+
+ CompletableFuture<Void> future = client.reportSchema(instance);
+
+ assertDoesNotThrow(() ->
+ future.whenComplete((ignored, throwable) -> assertNull(throwable))
+ .join());
Review Comment:
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "frankgh (via GitHub)" <gi...@apache.org>.
frankgh merged PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "frankgh (via GitHub)" <gi...@apache.org>.
frankgh commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1964011622
##########
server/src/main/java/org/apache/cassandra/sidecar/server/MainModule.java:
##########
@@ -598,6 +600,14 @@ public Router vertxRouter(Vertx vertx,
.handler(streamCdcSegmentHandler)
.build();
+ // Schema Reporting
+ protectedRouteBuilderFactory.get()
+ .router(router)
+ .method(HttpMethod.GET)
Review Comment:
The verb should not be GET. This should either be PUT or POST:
- PUT is more suitable for idempotent operations
- POST is more suitable for operations that are not idempotent
Also, this endpoint might be suitable for the operations framework. Something to consider
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1989716131
##########
server/src/test/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandlerTest.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import com.datastax.driver.core.Metadata;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Provides;
+import com.google.inject.Singleton;
+import com.google.inject.util.Modules;
+import com.linkedin.mxe.MetadataChangeProposal;
+import datahub.client.Callback;
+import datahub.client.MetadataWriteResponse;
+import datahub.shaded.findbugs.annotations.SuppressFBWarnings;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.Vertx;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.client.predicate.ResponsePredicate;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+import org.apache.cassandra.sidecar.TestModule;
+import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
+import org.apache.cassandra.sidecar.cluster.InstancesMetadata;
+import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.utils.IOUtils;
+import org.apache.cassandra.sidecar.datahub.EmitterFactory;
+import org.apache.cassandra.sidecar.datahub.JsonEmitter;
+import org.apache.cassandra.sidecar.server.MainModule;
+import org.apache.cassandra.sidecar.server.Server;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link ReportSchemaHandler}
+ */
+@ExtendWith(VertxExtension.class)
+final class ReportSchemaHandlerTest
+{
+ private static final String CLUSTER = "cluster";
+ private static final String DIRECTORY = "/tmp";
+ private static final int IDENTIFIER = 42;
+ private static final String LOCALHOST = "127.0.0.1";
+ private static final int PORT = 9042;
+ private static final String ENDPOINT = "/api/v1/report-schema";
+ private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+ private static final class ThrowingEmitter extends JsonEmitter
+ {
+ @Override
+ @NotNull
+ public synchronized Future<MetadataWriteResponse> emit(@NotNull MetadataChangeProposal proposal,
+ @Nullable Callback callback) throws IOException
+ {
+ throw new IOException();
+ }
+ }
+
+ private final class ReportSchemaHandlerTestModule extends AbstractModule
+ {
+ @Provides
+ @Singleton
+ @NotNull
+ public InstancesMetadata instancesMetadata()
+ {
+ Metadata metadata = mock(Metadata.class);
+ when(metadata.getKeyspaces()).thenReturn(Collections.emptyList());
+
+ StorageOperations operations = mock(StorageOperations.class);
+ when(operations.clusterName()).thenReturn(CLUSTER);
+
+ CassandraAdapterDelegate delegate = mock(CassandraAdapterDelegate.class);
+ when(delegate.storageOperations()).thenReturn(operations);
+ when(delegate.metadata()).thenReturn(metadata);
+
+ InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
+ when(instanceMetadata.stagingDir()).thenReturn(DIRECTORY);
+ when(instanceMetadata.id()).thenReturn(IDENTIFIER);
+ when(instanceMetadata.host()).thenReturn(LOCALHOST);
+ when(instanceMetadata.port()).thenReturn(PORT);
+ when(instanceMetadata.delegate()).thenReturn(delegate);
+
+ InstancesMetadata instances = mock(InstancesMetadata.class);
+ when(instances.instances()).thenReturn(Collections.singletonList(instanceMetadata));
+ when(instances.instanceFromId(IDENTIFIER)).thenReturn(instanceMetadata);
+ when(instances.instanceFromHost(LOCALHOST)).thenReturn(instanceMetadata);
+ return instances;
+ }
+
+ @Provides
+ @Singleton
+ @NotNull
+ public EmitterFactory emitterFactory()
+ {
+ return () -> emitter;
+ }
+ }
+
+ private final Injector injector = Guice.createInjector(Modules.override(new MainModule()).with(
+ Modules.override(new TestModule()).with(
+ new ReportSchemaHandlerTestModule())));
+ private WebClient client;
+ private Server server;
+ private JsonEmitter emitter;
+
+ @BeforeEach
+ void before() throws InterruptedException
+ {
+ VertxTestContext context = new VertxTestContext();
+ server = injector.getInstance(Server.class);
+ server.start()
+ .onSuccess(result -> context.completeNow())
+ .onFailure(context::failNow);
+
+ client = WebClient.create(injector.getInstance(Vertx.class));
+
+ context.awaitCompletion(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @AfterEach
+ @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED")
+ void after() throws InterruptedException
+ {
+ CountDownLatch latch = new CountDownLatch(1);
+ server.close()
+ .onSuccess(future -> latch.countDown());
+
+ client.close();
+
+ latch.await(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @Test
+ void testSuccess(@NotNull VertxTestContext context) throws IOException
+ {
+ String expected = IOUtils.readFully("/datahub/empty_cluster.json");
+ emitter = new JsonEmitter();
+ assertThat(emitter.content().length())
+ .isLessThanOrEqualTo(1);
+
+ client.put(server.actualPort(), LOCALHOST, ENDPOINT)
+ .expect(ResponsePredicate.SC_OK)
+ .send(context.succeeding(response ->
+ {
+ assertThat(response.statusCode())
+ .isEqualTo(HttpResponseStatus.OK.code());
+ assertThat(emitter.content())
+ .isEqualTo(expected);
+ context.completeNow();
+ }));
Review Comment:
Just to clarify.
1. run `./gradlew idea` to _apply_ the codestyle settings. It does not reformat, but update your project settings.
2. reformat on _all the updated files in this patch_
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1988201976
##########
client/src/testFixtures/java/org/apache/cassandra/sidecar/client/SidecarClientTest.java:
##########
@@ -1673,6 +1677,42 @@ public void onError(Throwable throwable)
assertThat(new String(baos.toByteArray(), StandardCharsets.UTF_8)).isEqualTo("Test Content");
}
+ @Test
+ public void testReportSchemaSuccess()
+ {
+ MockResponse response = new MockResponse()
+ .setResponseCode(OK.code())
+ .setBody("");
+
+ enqueue(response);
+
+ SidecarInstance instance = instances.get(0);
+
+ CompletableFuture<Void> future = client.reportSchema(instance);
+
+ assertDoesNotThrow(() ->
+ future.whenComplete((ignored, throwable) -> assertNull(throwable))
+ .join());
+ }
+
+ @Test
+ public void testReportSchemaFailure()
+ {
+ MockResponse response = new MockResponse()
+ .setResponseCode(INTERNAL_SERVER_ERROR.code())
+ .setBody("Message");
+
+ enqueue(response);
+
+ SidecarInstance instance = instances.get(0);
+
+ CompletableFuture<Void> future = client.reportSchema(instance);
+
+ assertThrows(CompletionException.class, () ->
+ future.whenComplete((ignored, throwable) -> assertNotNull(throwable))
+ .join());
Review Comment:
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1988209774
##########
server/src/main/java/org/apache/cassandra/sidecar/datahub/SchemaReporter.java:
##########
@@ -113,13 +113,23 @@ protected SchemaReporter(@NotNull IdentifiersProvider identifiersProvider,
/**
* Public method for converting and reporting the Cassandra schema
*
- * @param cluster a {@link Cluster} to extract Cassandra schema from
+ * @param cluster the {@link Cluster} to extract Cassandra schema from
*/
public void process(@NotNull Cluster cluster)
+ {
+ process(cluster.getMetadata());
+ }
+
+ /**
+ * Public method for converting and reporting the Cassandra schema
+ *
+ * @param metadata the {@link Metadata} to extract Cassandra schema from
+ */
+ public void process(@NotNull Metadata metadata)
{
try (Emitter emitter = emitterFactory.emitter())
{
- stream(cluster.getMetadata())
+ stream(metadata)
.forEach(ThrowableUtils.consumer(emitter::emit));
Review Comment:
It's way easier to read, discuss, *and* put breakpoints in this way, can we keep it as is, if not violating any explicit convention?
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1988209774
##########
server/src/main/java/org/apache/cassandra/sidecar/datahub/SchemaReporter.java:
##########
@@ -113,13 +113,23 @@ protected SchemaReporter(@NotNull IdentifiersProvider identifiersProvider,
/**
* Public method for converting and reporting the Cassandra schema
*
- * @param cluster a {@link Cluster} to extract Cassandra schema from
+ * @param cluster the {@link Cluster} to extract Cassandra schema from
*/
public void process(@NotNull Cluster cluster)
+ {
+ process(cluster.getMetadata());
+ }
+
+ /**
+ * Public method for converting and reporting the Cassandra schema
+ *
+ * @param metadata the {@link Metadata} to extract Cassandra schema from
+ */
+ public void process(@NotNull Metadata metadata)
{
try (Emitter emitter = emitterFactory.emitter())
{
- stream(cluster.getMetadata())
+ stream(metadata)
.forEach(ThrowableUtils.consumer(emitter::emit));
Review Comment:
It's way easier to read *and* put breakpoints in this way, can we keep it as is, if not violating any explicit convention?
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "frankgh (via GitHub)" <gi...@apache.org>.
frankgh commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1987628220
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandler.java:
##########
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.util.Collections;
+import java.util.Set;
+
+import com.datastax.driver.core.Metadata;
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.auth.authorization.Authorization;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.acl.authorization.BasicPermissions;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.datahub.SchemaReporter;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * An implementation of {@link AbstractHandler} used to trigger an immediate,
+ * synchronous conversion and report of the current schema
+ */
+@Singleton
+public class ReportSchemaHandler extends AbstractHandler<Void> implements AccessProtected
+{
+ @NotNull
+ private final SchemaReporter schemaReporter;
+
+ /**
+ * Constructs a new instance of {@link ReportSchemaHandler} using the provided instances
+ * of {@link InstanceMetadataFetcher}, {@link ExecutorPools}, and {@link SchemaReporter}
+ *
+ * @param metadata the metadata fetcher
+ * @param executor executor pools for blocking executions
+ * @param reporter schema reporter to use for the conversion
+ */
+ @Inject
+ public ReportSchemaHandler(@NotNull InstanceMetadataFetcher metadata,
+ @NotNull ExecutorPools executor,
+ @NotNull SchemaReporter reporter)
+ {
+ super(metadata, executor, null);
+
+ schemaReporter = reporter;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ @NotNull
+ public Set<Authorization> requiredAuthorizations()
+ {
+ return Collections.singleton(BasicPermissions.REPORT_SCHEMA.toAuthorization());
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ @Nullable
+ protected Void extractParamsOrThrow(@NotNull RoutingContext context)
+ {
+ return null;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected void handleInternal(@NotNull RoutingContext context,
+ @NotNull HttpServerRequest http,
+ @NotNull String host,
+ @NotNull SocketAddress address,
+ @Nullable Void request)
+ {
+ Metadata metadata = metadataFetcher.callOnFirstAvailableInstance(instance -> instance.delegate().metadata());
+
+ executorPools.service()
+ .runBlocking(() -> schemaReporter.process(metadata))
+ .onSuccess(ignored -> context.end())
Review Comment:
can we respond with a JSON response here? The expectation is that all endpoints return a JSON payload (except for streaming endpoints). I think we can do OK_STATUS
```suggestion
.onSuccess(ignored -> context.json(OK_STATUS))
```
##########
CHANGES.txt:
##########
Review Comment:
Looks like there is a merge conflict here.
##########
server/src/test/java/org/apache/cassandra/sidecar/utils/InstanceMetadataFetcherTest.java:
##########
Review Comment:
this entire class should be reverted before merge. Changes are unrelated to the PR.
##########
client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClient.java:
##########
@@ -567,6 +566,28 @@ public void streamCdcSegments(SidecarInstance sidecarInstance,
.build(), streamConsumer);
}
+ /**
+ * Sends a request to trigger an immediate, synchronous schema
+ * conversion and report on the specified instance of the Sidecar
+ * regardless of the periodic task schedule or status
+ *
+ * @param instance the {@link SidecarInstance} to receive the request
+ * @return a {@link CompletableFuture} for the request
+ */
+ public CompletableFuture<Void> reportSchema(SidecarInstance instance)
+ {
+ // Create an instance of {@link RequestContext.Builder} using its
+ // constructor instead of the {@link this.requestBuilder()} method,
+ // since {@link NoRetryPolicy} is the preferred behavior here
+
+ RequestContext context = new RequestContext.Builder()
+ .singleInstanceSelectionPolicy(instance)
+ .reportSchemaRequest()
+ .build();
+
+ return executor.executeRequestAsync(context);
Review Comment:
you can always override the policy like this:
```suggestion
return executor.executeRequestAsync(requestBuilder()
.singleInstanceSelectionPolicy(instance)
.retryPolicy(new NoRetryPolicy())
.reportSchemaRequest()
.build());
```
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/ApiEndpointsV1.java:
##########
@@ -121,13 +121,16 @@ public final class ApiEndpointsV1
public static final String LIST_CDC_SEGMENTS_ROUTE = API_V1 + CDC_PATH + "/segments";
public static final String STREAM_CDC_SEGMENTS_ROUTE = LIST_CDC_SEGMENTS_ROUTE + "/" + SEGMENT_PATH_PARAM;
+ // Schema Reporting
+ private static final String REPORT_SCHEMA = "/report-schema";
+ public static final String REPORT_SCHEMA_ROUTE = API_V1 + REPORT_SCHEMA;
Review Comment:
I'm not 100% convinced about this route. What do you think about this?
```suggestion
public static final String REPORT_SCHEMA_ROUTE = API_V1 + "/services/schema-reporting/ REPORT_SCHEMA;
```
or something that scopes this feature to the schema reporting service.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1964068967
##########
server/src/main/java/org/apache/cassandra/sidecar/acl/authorization/BasicPermissions.java:
##########
@@ -63,6 +63,9 @@ public class BasicPermissions
public static final Permission READ_OPERATIONAL_JOB = new DomainAwarePermission("OPERATIONAL_JOB:READ", OPERATION_SCOPE);
public static final Permission DECOMMISSION_NODE = new DomainAwarePermission("NODE:DECOMMISSION", OPERATION_SCOPE);
+ // Permissions related to Schema Reporting
+ public static final Permission REPORT_SCHEMA = new DomainAwarePermission("SCHEMA:REPORT", CLUSTER_SCOPE);
Review Comment:
The sentiment here is to have restraint in adding new verbs. Ideally, it should be a fixed set of verb to avoid operational pain.
The reason that READ should work here is that the reporter is reading the cassandra schema. When it publishes (i.e. sends requests to DataHub), the authorization should be enforced by the server (DataHub), not the client (Sidecar).
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1984014570
##########
server/src/main/java/org/apache/cassandra/sidecar/utils/InstanceMetadataFetcher.java:
##########
@@ -97,13 +97,29 @@ public CassandraAdapterDelegate delegate(@NotNull String host) throws NoSuchCass
}
/**
- * Iterate through the local instances and call the function on the first available instance, i.e. no CassandraUnavailableException
- * or OperationUnavailableException is thrown for the operations
+ * Iterate through the local instances and run the {@link Consumer} on the first available one,
+ * so no {@link CassandraUnavailableException} or {@link OperationUnavailableException} is thrown for the operations
+ *
+ * @param consumer a {@link Consumer} that processes {@link InstanceMetadata} and returns no result
+ * @throws CassandraUnavailableException if all local instances were exhausted
+ */
+ public void runOnFirstAvailableInstance(Consumer<InstanceMetadata> consumer) throws CassandraUnavailableException
+ {
+ callOnFirstAvailableInstance(metadata ->
+ {
+ consumer.accept(metadata);
+ return null;
+ });
+ }
Review Comment:
Removed the method, tests, stylistic changes.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1987740739
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/ApiEndpointsV1.java:
##########
@@ -121,13 +121,16 @@ public final class ApiEndpointsV1
public static final String LIST_CDC_SEGMENTS_ROUTE = API_V1 + CDC_PATH + "/segments";
public static final String STREAM_CDC_SEGMENTS_ROUTE = LIST_CDC_SEGMENTS_ROUTE + "/" + SEGMENT_PATH_PARAM;
+ // Schema Reporting
+ private static final String REPORT_SCHEMA = "/report-schema";
+ public static final String REPORT_SCHEMA_ROUTE = API_V1 + REPORT_SCHEMA;
Review Comment:
I agree. The API should not contain verb. Given the API is to interact with DataHub, I would suggest the following to explicitly refer to DataHub
```
PUT /api/v1/datahub/schemas
```
Meaning, update the schema in datahub.
The API on the datahub side is `/openapi/entities/v1`. Maybe `entities` also work if we want to define a similar API. However, I think `schemas` provides better clarity in the context of Sidecar.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1987945295
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/ApiEndpointsV1.java:
##########
@@ -121,13 +121,16 @@ public final class ApiEndpointsV1
public static final String LIST_CDC_SEGMENTS_ROUTE = API_V1 + CDC_PATH + "/segments";
public static final String STREAM_CDC_SEGMENTS_ROUTE = LIST_CDC_SEGMENTS_ROUTE + "/" + SEGMENT_PATH_PARAM;
+ // Schema Reporting
+ private static final String REPORT_SCHEMA = "/report-schema";
+ public static final String REPORT_SCHEMA_ROUTE = API_V1 + REPORT_SCHEMA;
Review Comment:
I would agree that `PUT .../schema` (singular, not plural) is an appropriate name if we provide the schema we intent to persist. But we're not putting schema in. We're triggering an event that causes the schema to be internally produced and reported. Using `PUT .../schema-report` or `PUT .../schema-report-request` would probably follow the general idea of REST better. What do you think?
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1988206374
##########
client/src/testFixtures/java/org/apache/cassandra/sidecar/client/SidecarClientTest.java:
##########
@@ -1673,6 +1677,42 @@ public void onError(Throwable throwable)
assertThat(new String(baos.toByteArray(), StandardCharsets.UTF_8)).isEqualTo("Test Content");
}
+ @Test
+ public void testReportSchemaSuccess()
+ {
+ MockResponse response = new MockResponse()
+ .setResponseCode(OK.code())
+ .setBody("");
+
+ enqueue(response);
+
+ SidecarInstance instance = instances.get(0);
+
+ CompletableFuture<Void> future = client.reportSchema(instance);
+
+ assertDoesNotThrow(() ->
+ future.whenComplete((ignored, throwable) -> assertNull(throwable))
+ .join());
+ }
+
+ @Test
+ public void testReportSchemaFailure()
+ {
+ MockResponse response = new MockResponse()
+ .setResponseCode(INTERNAL_SERVER_ERROR.code())
+ .setBody("Message");
+
+ enqueue(response);
+
+ SidecarInstance instance = instances.get(0);
+
+ CompletableFuture<Void> future = client.reportSchema(instance);
+
+ assertThrows(CompletionException.class, () ->
+ future.whenComplete((ignored, throwable) -> assertNotNull(throwable))
+ .join());
+ }
+
Review Comment:
I've changed these a lot, let me know if the formatting still needs change.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1987923135
##########
client/src/main/java/org/apache/cassandra/sidecar/client/SidecarClient.java:
##########
@@ -567,6 +566,28 @@ public void streamCdcSegments(SidecarInstance sidecarInstance,
.build(), streamConsumer);
}
+ /**
+ * Sends a request to trigger an immediate, synchronous schema
+ * conversion and report on the specified instance of the Sidecar
+ * regardless of the periodic task schedule or status
+ *
+ * @param instance the {@link SidecarInstance} to receive the request
+ * @return a {@link CompletableFuture} for the request
+ */
+ public CompletableFuture<Void> reportSchema(SidecarInstance instance)
+ {
+ // Create an instance of {@link RequestContext.Builder} using its
+ // constructor instead of the {@link this.requestBuilder()} method,
+ // since {@link NoRetryPolicy} is the preferred behavior here
+
+ RequestContext context = new RequestContext.Builder()
+ .singleInstanceSelectionPolicy(instance)
+ .reportSchemaRequest()
+ .build();
+
+ return executor.executeRequestAsync(context);
Review Comment:
Added a convenience method for policy overrides following the existing pattern.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "sarankk (via GitHub)" <gi...@apache.org>.
sarankk commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1964164599
##########
server/src/main/java/org/apache/cassandra/sidecar/acl/authorization/BasicPermissions.java:
##########
@@ -63,6 +63,9 @@ public class BasicPermissions
public static final Permission READ_OPERATIONAL_JOB = new DomainAwarePermission("OPERATIONAL_JOB:READ", OPERATION_SCOPE);
public static final Permission DECOMMISSION_NODE = new DomainAwarePermission("NODE:DECOMMISSION", OPERATION_SCOPE);
+ // Permissions related to Schema Reporting
+ public static final Permission REPORT_SCHEMA = new DomainAwarePermission("SCHEMA:REPORT", CLUSTER_SCOPE);
Review Comment:
I feel having 2 different permissions is better in general, with 1 granting read permission automatically grants them report permission. But in this particular case, since we already have a periodic task to report schema, irrespective of this endpoint, we can use the `enable` flag for schema reporting to control whether we want to allow reporting or not without adding a separate permission for reporting control?
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1970471506
##########
CHANGES.txt:
##########
@@ -8,7 +9,7 @@
* Sidecar schema initialization can be executed on multiple thread (CASSSIDECAR-200)
* Make sidecar operations resilient to down Cassandra nodes (CASSSIDECAR-201)
* Fix Cassandra instance not found error (CASSSIDECAR-192)
- * Implemented Schema Reporter for Integration with DataHub (CASSSIDECAR-191)
+ * Implement Schema Reporter for Integration with DataHub (CASSSIDECAR-191)
Review Comment:
It is mainly avoid amending the changes log. It is not really a typo. "Implemented something" described the change as clear.
If you look at the log, there are places where "Adds" and "Add" are used. There is no strict rule on the verb's form.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1987764596
##########
client/src/testFixtures/java/org/apache/cassandra/sidecar/client/SidecarClientTest.java:
##########
@@ -1673,6 +1677,42 @@ public void onError(Throwable throwable)
assertThat(new String(baos.toByteArray(), StandardCharsets.UTF_8)).isEqualTo("Test Content");
}
+ @Test
+ public void testReportSchemaSuccess()
+ {
+ MockResponse response = new MockResponse()
+ .setResponseCode(OK.code())
+ .setBody("");
+
+ enqueue(response);
+
+ SidecarInstance instance = instances.get(0);
+
+ CompletableFuture<Void> future = client.reportSchema(instance);
+
+ assertDoesNotThrow(() ->
+ future.whenComplete((ignored, throwable) -> assertNull(throwable))
+ .join());
+ }
+
+ @Test
+ public void testReportSchemaFailure()
+ {
+ MockResponse response = new MockResponse()
+ .setResponseCode(INTERNAL_SERVER_ERROR.code())
+ .setBody("Message");
+
+ enqueue(response);
+
+ SidecarInstance instance = instances.get(0);
+
+ CompletableFuture<Void> future = client.reportSchema(instance);
+
+ assertThrows(CompletionException.class, () ->
+ future.whenComplete((ignored, throwable) -> assertNotNull(throwable))
+ .join());
Review Comment:
Use assertJ and enrich the assertion
```suggestion
SidecarInstance instance = instances.get(0);
assertThatThrownBy(() -> client.reportSchema(instance).get())
.isExactlyInstanceOf(ExecutionException.class)
.hasCauseInstanceOf(RetriesExhaustedException.class)
.hasMessageContaining("statusCode=500, statusMessage='Server Error'");
```
##########
client/src/testFixtures/java/org/apache/cassandra/sidecar/client/SidecarClientTest.java:
##########
@@ -1673,6 +1677,42 @@ public void onError(Throwable throwable)
assertThat(new String(baos.toByteArray(), StandardCharsets.UTF_8)).isEqualTo("Test Content");
}
+ @Test
+ public void testReportSchemaSuccess()
+ {
+ MockResponse response = new MockResponse()
+ .setResponseCode(OK.code())
+ .setBody("");
+
+ enqueue(response);
+
+ SidecarInstance instance = instances.get(0);
+
+ CompletableFuture<Void> future = client.reportSchema(instance);
+
+ assertDoesNotThrow(() ->
+ future.whenComplete((ignored, throwable) -> assertNull(throwable))
+ .join());
+ }
+
+ @Test
+ public void testReportSchemaFailure()
+ {
+ MockResponse response = new MockResponse()
+ .setResponseCode(INTERNAL_SERVER_ERROR.code())
+ .setBody("Message");
+
+ enqueue(response);
+
+ SidecarInstance instance = instances.get(0);
+
+ CompletableFuture<Void> future = client.reportSchema(instance);
+
+ assertThrows(CompletionException.class, () ->
+ future.whenComplete((ignored, throwable) -> assertNotNull(throwable))
+ .join());
+ }
+
Review Comment:
formatting in this new tests
##########
client/src/testFixtures/java/org/apache/cassandra/sidecar/client/SidecarClientTest.java:
##########
@@ -1673,6 +1677,42 @@ public void onError(Throwable throwable)
assertThat(new String(baos.toByteArray(), StandardCharsets.UTF_8)).isEqualTo("Test Content");
}
+ @Test
+ public void testReportSchemaSuccess()
+ {
+ MockResponse response = new MockResponse()
+ .setResponseCode(OK.code())
+ .setBody("");
+
+ enqueue(response);
+
+ SidecarInstance instance = instances.get(0);
+
+ CompletableFuture<Void> future = client.reportSchema(instance);
+
+ assertDoesNotThrow(() ->
+ future.whenComplete((ignored, throwable) -> assertNull(throwable))
+ .join());
Review Comment:
use assertJ
```suggestion
SidecarInstance instance = instances.get(0);
assertThatNoException().isThrownBy(() -> client.reportSchema(instance).get());
```
##########
server/src/test/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandlerTest.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import com.datastax.driver.core.Metadata;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Provides;
+import com.google.inject.Singleton;
+import com.google.inject.util.Modules;
+import com.linkedin.mxe.MetadataChangeProposal;
+import datahub.client.Callback;
+import datahub.client.MetadataWriteResponse;
+import datahub.shaded.findbugs.annotations.SuppressFBWarnings;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.Vertx;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.client.predicate.ResponsePredicate;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+import org.apache.cassandra.sidecar.TestModule;
+import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
+import org.apache.cassandra.sidecar.cluster.InstancesMetadata;
+import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.utils.IOUtils;
+import org.apache.cassandra.sidecar.datahub.EmitterFactory;
+import org.apache.cassandra.sidecar.datahub.JsonEmitter;
+import org.apache.cassandra.sidecar.server.MainModule;
+import org.apache.cassandra.sidecar.server.Server;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link ReportSchemaHandler}
+ */
+@ExtendWith(VertxExtension.class)
+final class ReportSchemaHandlerTest
+{
+ private static final String CLUSTER = "cluster";
+ private static final String DIRECTORY = "/tmp";
+ private static final int IDENTIFIER = 42;
+ private static final String LOCALHOST = "127.0.0.1";
+ private static final int PORT = 9042;
+ private static final String ENDPOINT = "/api/v1/report-schema";
+ private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+ private static final class ThrowingEmitter extends JsonEmitter
+ {
+ @Override
+ @NotNull
+ public synchronized Future<MetadataWriteResponse> emit(@NotNull MetadataChangeProposal proposal,
+ @Nullable Callback callback) throws IOException
+ {
+ throw new IOException();
+ }
+ }
+
+ private final class ReportSchemaHandlerTestModule extends AbstractModule
+ {
+ @Provides
+ @Singleton
+ @NotNull
+ public InstancesMetadata instancesMetadata()
+ {
+ Metadata metadata = mock(Metadata.class);
+ when(metadata.getKeyspaces()).thenReturn(Collections.emptyList());
+
+ StorageOperations operations = mock(StorageOperations.class);
+ when(operations.clusterName()).thenReturn(CLUSTER);
+
+ CassandraAdapterDelegate delegate = mock(CassandraAdapterDelegate.class);
+ when(delegate.storageOperations()).thenReturn(operations);
+ when(delegate.metadata()).thenReturn(metadata);
+
+ InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
+ when(instanceMetadata.stagingDir()).thenReturn(DIRECTORY);
+ when(instanceMetadata.id()).thenReturn(IDENTIFIER);
+ when(instanceMetadata.host()).thenReturn(LOCALHOST);
+ when(instanceMetadata.port()).thenReturn(PORT);
+ when(instanceMetadata.delegate()).thenReturn(delegate);
+
+ InstancesMetadata instances = mock(InstancesMetadata.class);
+ when(instances.instances()).thenReturn(Collections.singletonList(instanceMetadata));
+ when(instances.instanceFromId(IDENTIFIER)).thenReturn(instanceMetadata);
+ when(instances.instanceFromHost(LOCALHOST)).thenReturn(instanceMetadata);
+ return instances;
+ }
+
+ @Provides
+ @Singleton
+ @NotNull
+ public EmitterFactory emitterFactory()
+ {
+ return () -> emitter;
+ }
+ }
+
+ private final Injector injector = Guice.createInjector(Modules.override(new MainModule()).with(
+ Modules.override(new TestModule()).with(
+ new ReportSchemaHandlerTestModule())));
+ private WebClient client;
+ private Server server;
+ private JsonEmitter emitter;
+
+ @BeforeEach
+ void before() throws InterruptedException
+ {
+ VertxTestContext context = new VertxTestContext();
+ server = injector.getInstance(Server.class);
+ server.start()
+ .onSuccess(result -> context.completeNow())
+ .onFailure(context::failNow);
+
+ client = WebClient.create(injector.getInstance(Vertx.class));
+
+ context.awaitCompletion(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @AfterEach
+ @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED")
+ void after() throws InterruptedException
+ {
+ CountDownLatch latch = new CountDownLatch(1);
+ server.close()
+ .onSuccess(future -> latch.countDown());
+
+ client.close();
+
+ latch.await(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @Test
+ void testSuccess(@NotNull VertxTestContext context) throws IOException
+ {
+ String expected = IOUtils.readFully("/datahub/empty_cluster.json");
+ emitter = new JsonEmitter();
+ assertThat(emitter.content().length())
+ .isLessThanOrEqualTo(1);
+
+ client.put(server.actualPort(), LOCALHOST, ENDPOINT)
+ .expect(ResponsePredicate.SC_OK)
+ .send(context.succeeding(response ->
+ {
+ assertThat(response.statusCode())
+ .isEqualTo(HttpResponseStatus.OK.code());
+ assertThat(emitter.content())
+ .isEqualTo(expected);
+ context.completeNow();
+ }));
Review Comment:
Formatting.
Can you please run `./gradlew idea` to apply the codestyle settings and reformat on all the updated files in this patch? The `CONTINUATION_INDENT_SIZE` is 0.
##########
server/src/main/java/org/apache/cassandra/sidecar/datahub/SchemaReporter.java:
##########
@@ -113,13 +113,23 @@ protected SchemaReporter(@NotNull IdentifiersProvider identifiersProvider,
/**
* Public method for converting and reporting the Cassandra schema
*
- * @param cluster a {@link Cluster} to extract Cassandra schema from
+ * @param cluster the {@link Cluster} to extract Cassandra schema from
*/
public void process(@NotNull Cluster cluster)
+ {
+ process(cluster.getMetadata());
+ }
+
+ /**
+ * Public method for converting and reporting the Cassandra schema
+ *
+ * @param metadata the {@link Metadata} to extract Cassandra schema from
+ */
+ public void process(@NotNull Metadata metadata)
{
try (Emitter emitter = emitterFactory.emitter())
{
- stream(cluster.getMetadata())
+ stream(metadata)
.forEach(ThrowableUtils.consumer(emitter::emit));
Review Comment:
```suggestion
stream(metadata).forEach(ThrowableUtils.consumer(emitter::emit));
```
##########
server/src/test/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandlerTest.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import com.datastax.driver.core.Metadata;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Provides;
+import com.google.inject.Singleton;
+import com.google.inject.util.Modules;
+import com.linkedin.mxe.MetadataChangeProposal;
+import datahub.client.Callback;
+import datahub.client.MetadataWriteResponse;
+import datahub.shaded.findbugs.annotations.SuppressFBWarnings;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.Vertx;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.client.predicate.ResponsePredicate;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+import org.apache.cassandra.sidecar.TestModule;
+import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
+import org.apache.cassandra.sidecar.cluster.InstancesMetadata;
+import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.utils.IOUtils;
+import org.apache.cassandra.sidecar.datahub.EmitterFactory;
+import org.apache.cassandra.sidecar.datahub.JsonEmitter;
+import org.apache.cassandra.sidecar.server.MainModule;
+import org.apache.cassandra.sidecar.server.Server;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link ReportSchemaHandler}
+ */
+@ExtendWith(VertxExtension.class)
+final class ReportSchemaHandlerTest
+{
+ private static final String CLUSTER = "cluster";
+ private static final String DIRECTORY = "/tmp";
+ private static final int IDENTIFIER = 42;
+ private static final String LOCALHOST = "127.0.0.1";
+ private static final int PORT = 9042;
+ private static final String ENDPOINT = "/api/v1/report-schema";
+ private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+ private static final class ThrowingEmitter extends JsonEmitter
+ {
+ @Override
+ @NotNull
+ public synchronized Future<MetadataWriteResponse> emit(@NotNull MetadataChangeProposal proposal,
+ @Nullable Callback callback) throws IOException
+ {
+ throw new IOException();
+ }
+ }
+
+ private final class ReportSchemaHandlerTestModule extends AbstractModule
+ {
+ @Provides
+ @Singleton
+ @NotNull
+ public InstancesMetadata instancesMetadata()
+ {
+ Metadata metadata = mock(Metadata.class);
+ when(metadata.getKeyspaces()).thenReturn(Collections.emptyList());
+
+ StorageOperations operations = mock(StorageOperations.class);
+ when(operations.clusterName()).thenReturn(CLUSTER);
+
+ CassandraAdapterDelegate delegate = mock(CassandraAdapterDelegate.class);
+ when(delegate.storageOperations()).thenReturn(operations);
+ when(delegate.metadata()).thenReturn(metadata);
+
+ InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
+ when(instanceMetadata.stagingDir()).thenReturn(DIRECTORY);
+ when(instanceMetadata.id()).thenReturn(IDENTIFIER);
+ when(instanceMetadata.host()).thenReturn(LOCALHOST);
+ when(instanceMetadata.port()).thenReturn(PORT);
+ when(instanceMetadata.delegate()).thenReturn(delegate);
+
+ InstancesMetadata instances = mock(InstancesMetadata.class);
+ when(instances.instances()).thenReturn(Collections.singletonList(instanceMetadata));
+ when(instances.instanceFromId(IDENTIFIER)).thenReturn(instanceMetadata);
+ when(instances.instanceFromHost(LOCALHOST)).thenReturn(instanceMetadata);
+ return instances;
+ }
+
+ @Provides
+ @Singleton
+ @NotNull
+ public EmitterFactory emitterFactory()
+ {
+ return () -> emitter;
+ }
+ }
+
+ private final Injector injector = Guice.createInjector(Modules.override(new MainModule()).with(
+ Modules.override(new TestModule()).with(
+ new ReportSchemaHandlerTestModule())));
+ private WebClient client;
+ private Server server;
+ private JsonEmitter emitter;
+
+ @BeforeEach
+ void before() throws InterruptedException
+ {
+ VertxTestContext context = new VertxTestContext();
+ server = injector.getInstance(Server.class);
+ server.start()
+ .onSuccess(result -> context.completeNow())
+ .onFailure(context::failNow);
+
+ client = WebClient.create(injector.getInstance(Vertx.class));
+
+ context.awaitCompletion(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @AfterEach
+ @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED")
+ void after() throws InterruptedException
+ {
+ CountDownLatch latch = new CountDownLatch(1);
+ server.close()
+ .onSuccess(future -> latch.countDown());
+
+ client.close();
+
+ latch.await(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
Review Comment:
```suggestion
TestResourceReaper.create().with(server).with(client).close();
```
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1984012207
##########
CHANGES.txt:
##########
@@ -8,7 +9,7 @@
* Sidecar schema initialization can be executed on multiple thread (CASSSIDECAR-200)
* Make sidecar operations resilient to down Cassandra nodes (CASSSIDECAR-201)
* Fix Cassandra instance not found error (CASSSIDECAR-192)
- * Implemented Schema Reporter for Integration with DataHub (CASSSIDECAR-191)
+ * Implement Schema Reporter for Integration with DataHub (CASSSIDECAR-191)
Review Comment:
Got it, thanks for explaining, reverted the second chage.
(I was suspecting that maybe there's some kind of automation somewhere that depends on `blame` or something…)
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1984014907
##########
server/src/main/java/org/apache/cassandra/sidecar/server/MainModule.java:
##########
@@ -902,7 +912,8 @@ public IdentifiersProvider identifiersProvider(@NotNull InstanceMetadataFetcher
@NotNull
protected String initialize()
{
- return fetcher.callOnFirstAvailableInstance(i -> i.delegate().storageOperations().clusterName());
+ return fetcher.callOnFirstAvailableInstance(instance ->
+ instance.delegate().storageOperations().clusterName());
Review Comment:
Reverted.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1970471506
##########
CHANGES.txt:
##########
@@ -8,7 +9,7 @@
* Sidecar schema initialization can be executed on multiple thread (CASSSIDECAR-200)
* Make sidecar operations resilient to down Cassandra nodes (CASSSIDECAR-201)
* Fix Cassandra instance not found error (CASSSIDECAR-192)
- * Implemented Schema Reporter for Integration with DataHub (CASSSIDECAR-191)
+ * Implement Schema Reporter for Integration with DataHub (CASSSIDECAR-191)
Review Comment:
It is mainly avoid amending the changes log. It is not really a typo. "Implemented something" described the change as clear.
If you look at the log, there are places where "Adds" and "Add", as well as "Adding" are used. There is no strict rule on the verb's form.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1984012719
##########
server/src/main/java/org/apache/cassandra/sidecar/acl/authorization/BasicPermissions.java:
##########
@@ -63,6 +63,9 @@ public class BasicPermissions
public static final Permission READ_OPERATIONAL_JOB = new DomainAwarePermission("OPERATIONAL_JOB:READ", OPERATION_SCOPE);
public static final Permission DECOMMISSION_NODE = new DomainAwarePermission("NODE:DECOMMISSION", OPERATION_SCOPE);
+ // Permissions related to Schema Reporting
+ public static final Permission REPORT_SCHEMA = new DomainAwarePermission("SCHEMA:REPORT", CLUSTER_SCOPE);
Review Comment:
Changed the value to `SCHEMA:PUBLISH`, 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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1984018629
##########
server/src/test/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandlerTest.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import com.datastax.driver.core.Metadata;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Provides;
+import com.google.inject.Singleton;
+import com.google.inject.util.Modules;
+import com.linkedin.mxe.MetadataChangeProposal;
+import datahub.client.Callback;
+import datahub.client.MetadataWriteResponse;
+import datahub.shaded.findbugs.annotations.SuppressFBWarnings;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.Vertx;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.client.predicate.ResponsePredicate;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+import org.apache.cassandra.sidecar.TestModule;
+import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
+import org.apache.cassandra.sidecar.cluster.InstancesMetadata;
+import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.utils.IOUtils;
+import org.apache.cassandra.sidecar.datahub.EmitterFactory;
+import org.apache.cassandra.sidecar.datahub.JsonEmitter;
+import org.apache.cassandra.sidecar.server.MainModule;
+import org.apache.cassandra.sidecar.server.Server;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link ReportSchemaHandler}
+ */
+@ExtendWith(VertxExtension.class)
+final class ReportSchemaHandlerTest
+{
+ private static final String CLUSTER = "cluster";
+ private static final String DIRECTORY = "/tmp";
+ private static final int IDENTIFIER = 42;
+ private static final String LOCALHOST = "127.0.0.1";
+ private static final int PORT = 9042;
+ private static final String ENDPOINT = "/api/v1/report-schema";
+ private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+ private static final class ThrowingEmitter extends JsonEmitter
+ {
+ @Override
+ @NotNull
+ public synchronized Future<MetadataWriteResponse> emit(@NotNull MetadataChangeProposal proposal,
+ @Nullable Callback callback) throws IOException
+ {
+ throw new IOException();
+ }
+ }
+
+ private final class ReportSchemaHandlerTestModule extends AbstractModule
+ {
+ @Provides
+ @Singleton
+ @NotNull
+ public InstancesMetadata instancesMetadata()
+ {
+ Metadata metadata = mock(Metadata.class);
+ when(metadata.getKeyspaces()).thenReturn(Collections.emptyList());
+
+ StorageOperations operations = mock(StorageOperations.class);
+ when(operations.clusterName()).thenReturn(CLUSTER);
+
+ CassandraAdapterDelegate delegate = mock(CassandraAdapterDelegate.class);
+ when(delegate.storageOperations()).thenReturn(operations);
+ when(delegate.metadata()).thenReturn(metadata);
+
+ InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
+ when(instanceMetadata.stagingDir()).thenReturn(DIRECTORY);
+ when(instanceMetadata.id()).thenReturn(IDENTIFIER);
+ when(instanceMetadata.host()).thenReturn(LOCALHOST);
+ when(instanceMetadata.port()).thenReturn(PORT);
+ when(instanceMetadata.delegate()).thenReturn(delegate);
+
+ InstancesMetadata instances = mock(InstancesMetadata.class);
+ when(instances.instances()).thenReturn(Collections.singletonList(instanceMetadata));
+ when(instances.instanceFromId(IDENTIFIER)).thenReturn(instanceMetadata);
+ when(instances.instanceFromHost(LOCALHOST)).thenReturn(instanceMetadata);
+ return instances;
+ }
+
+ @Provides
+ @Singleton
+ @NotNull
+ public EmitterFactory emitterFactory()
+ {
+ return () -> emitter;
+ }
+ }
+
+ private final Injector injector = Guice.createInjector(Modules.override(new MainModule()).with(
+ Modules.override(new TestModule()).with(
+ new ReportSchemaHandlerTestModule())));
+ private WebClient client;
+ private Server server;
+ private JsonEmitter emitter;
+
+ @BeforeEach
+ void before() throws InterruptedException
+ {
+ client = WebClient.create(injector.getInstance(Vertx.class));
+ server = injector.getInstance(Server.class);
+
+ VertxTestContext context = new VertxTestContext();
+ server.start()
+ .onSuccess(result -> context.completeNow())
+ .onFailure(context::failNow);
+ context.awaitCompletion(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @AfterEach
+ @SuppressWarnings("ResultOfMethodCallIgnored")
+ @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED")
+ void after() throws InterruptedException
+ {
+ CountDownLatch latch = new CountDownLatch(1);
+ server.close()
+ .onSuccess(future -> latch.countDown());
+ latch.await(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @Test
+ @SuppressWarnings("deprecation")
+ void testSuccess(@NotNull VertxTestContext context) throws IOException
+ {
+ String expected = IOUtils.readFully("/datahub/empty_cluster.json");
+ emitter = new JsonEmitter();
+
+ client.get(server.actualPort(), LOCALHOST, ENDPOINT)
Review Comment:
Added a check (not empty per se, just an opening bracket).
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1984017857
##########
server/src/test/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandlerTest.java:
##########
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import com.datastax.driver.core.Metadata;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Provides;
+import com.google.inject.Singleton;
+import com.google.inject.util.Modules;
+import com.linkedin.mxe.MetadataChangeProposal;
+import datahub.client.Callback;
+import datahub.client.MetadataWriteResponse;
+import datahub.shaded.findbugs.annotations.SuppressFBWarnings;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.Vertx;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.client.predicate.ResponsePredicate;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+import org.apache.cassandra.sidecar.TestModule;
+import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
+import org.apache.cassandra.sidecar.cluster.InstancesMetadata;
+import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.utils.IOUtils;
+import org.apache.cassandra.sidecar.datahub.EmitterFactory;
+import org.apache.cassandra.sidecar.datahub.JsonEmitter;
+import org.apache.cassandra.sidecar.server.MainModule;
+import org.apache.cassandra.sidecar.server.Server;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link ReportSchemaHandler}
+ */
+@ExtendWith(VertxExtension.class)
+final class ReportSchemaHandlerTest
+{
+ private static final String CLUSTER = "cluster";
+ private static final String DIRECTORY = "/tmp";
+ private static final int IDENTIFIER = 42;
+ private static final String LOCALHOST = "127.0.0.1";
+ private static final int PORT = 9042;
+ private static final String ENDPOINT = "/api/v1/report-schema";
+ private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+ private static final class ThrowingEmitter extends JsonEmitter
+ {
+ @Override
+ @NotNull
+ public synchronized Future<MetadataWriteResponse> emit(@NotNull MetadataChangeProposal proposal,
+ @Nullable Callback callback) throws IOException
+ {
+ throw new IOException();
+ }
+ }
+
+ private final class ReportSchemaHandlerTestModule extends AbstractModule
+ {
+ @Provides
+ @Singleton
+ @NotNull
+ public InstancesMetadata instancesMetadata()
+ {
+ Metadata metadata = mock(Metadata.class);
+ when(metadata.getKeyspaces()).thenReturn(Collections.emptyList());
+
+ StorageOperations operations = mock(StorageOperations.class);
+ when(operations.clusterName()).thenReturn(CLUSTER);
+
+ CassandraAdapterDelegate delegate = mock(CassandraAdapterDelegate.class);
+ when(delegate.storageOperations()).thenReturn(operations);
+ when(delegate.metadata()).thenReturn(metadata);
+
+ InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
+ when(instanceMetadata.stagingDir()).thenReturn(DIRECTORY);
+ when(instanceMetadata.id()).thenReturn(IDENTIFIER);
+ when(instanceMetadata.host()).thenReturn(LOCALHOST);
+ when(instanceMetadata.port()).thenReturn(PORT);
+ when(instanceMetadata.delegate()).thenReturn(delegate);
+
+ InstancesMetadata instances = mock(InstancesMetadata.class);
+ when(instances.instances()).thenReturn(Collections.singletonList(instanceMetadata));
+ when(instances.instanceFromId(IDENTIFIER)).thenReturn(instanceMetadata);
+ when(instances.instanceFromHost(LOCALHOST)).thenReturn(instanceMetadata);
+ return instances;
+ }
+
+ @Provides
+ @Singleton
+ @NotNull
+ public EmitterFactory emitterFactory()
+ {
+ return () -> emitter;
+ }
+ }
+
+ private final Injector injector = Guice.createInjector(Modules.override(new MainModule()).with(
+ Modules.override(new TestModule()).with(
+ new ReportSchemaHandlerTestModule())));
+ private WebClient client;
+ private Server server;
+ private JsonEmitter emitter;
+
+ @BeforeEach
+ void before() throws InterruptedException
+ {
+ client = WebClient.create(injector.getInstance(Vertx.class));
+ server = injector.getInstance(Server.class);
+
+ VertxTestContext context = new VertxTestContext();
+ server.start()
+ .onSuccess(result -> context.completeNow())
+ .onFailure(context::failNow);
+ context.awaitCompletion(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @AfterEach
+ @SuppressWarnings("ResultOfMethodCallIgnored")
+ @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED")
+ void after() throws InterruptedException
+ {
+ CountDownLatch latch = new CountDownLatch(1);
+ server.close()
+ .onSuccess(future -> latch.countDown());
+ latch.await(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
Review Comment:
Added instruction to close the client, changed order to improve performance.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1984019295
##########
server/src/main/java/org/apache/cassandra/sidecar/server/MainModule.java:
##########
@@ -598,6 +600,14 @@ public Router vertxRouter(Vertx vertx,
.handler(streamCdcSegmentHandler)
.build();
+ // Schema Reporting
+ protectedRouteBuilderFactory.get()
+ .router(router)
+ .method(HttpMethod.GET)
Review Comment:
Good point, changed everywhere to `PUT`.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1987925373
##########
server/src/test/java/org/apache/cassandra/sidecar/utils/InstanceMetadataFetcherTest.java:
##########
Review Comment:
That was my intention, reverted the overlooked changes.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1964638954
##########
server/src/main/java/org/apache/cassandra/sidecar/acl/authorization/BasicPermissions.java:
##########
@@ -63,6 +63,9 @@ public class BasicPermissions
public static final Permission READ_OPERATIONAL_JOB = new DomainAwarePermission("OPERATIONAL_JOB:READ", OPERATION_SCOPE);
public static final Permission DECOMMISSION_NODE = new DomainAwarePermission("NODE:DECOMMISSION", OPERATION_SCOPE);
+ // Permissions related to Schema Reporting
+ public static final Permission REPORT_SCHEMA = new DomainAwarePermission("SCHEMA:REPORT", CLUSTER_SCOPE);
Review Comment:
Let's also think about the future. We do not want to go down on the route of introducing new verbs for new actions. And this is triggering my concern on it.
I am fine with letting `PUBLISH` pass. But we should really think about it and be mindful with introducing new verbs.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1964172369
##########
server/src/main/java/org/apache/cassandra/sidecar/acl/authorization/BasicPermissions.java:
##########
@@ -63,6 +63,9 @@ public class BasicPermissions
public static final Permission READ_OPERATIONAL_JOB = new DomainAwarePermission("OPERATIONAL_JOB:READ", OPERATION_SCOPE);
public static final Permission DECOMMISSION_NODE = new DomainAwarePermission("NODE:DECOMMISSION", OPERATION_SCOPE);
+ // Permissions related to Schema Reporting
+ public static final Permission REPORT_SCHEMA = new DomainAwarePermission("SCHEMA:REPORT", CLUSTER_SCOPE);
Review Comment:
I was planning to get to this tomorrow, but see there's confusion about this one.
Changing permissions to the existing `SCHEMA:READ` will be logically equivalent to the following statement:
> "Everyone who is allowed to see cluster schema is also allowed to perform DoS attacks on Sidecar."
Is that actually true?
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1970471506
##########
CHANGES.txt:
##########
@@ -8,7 +9,7 @@
* Sidecar schema initialization can be executed on multiple thread (CASSSIDECAR-200)
* Make sidecar operations resilient to down Cassandra nodes (CASSSIDECAR-201)
* Fix Cassandra instance not found error (CASSSIDECAR-192)
- * Implemented Schema Reporter for Integration with DataHub (CASSSIDECAR-191)
+ * Implement Schema Reporter for Integration with DataHub (CASSSIDECAR-191)
Review Comment:
It is mainly avoid amending the changes log. It is more important to have the git history of each line linked to the correct commit.
Btw, it is not really a typo. "Implemented something" described the change as clear.
If you look at the log, there are places where "Adds" and "Add", as well as "Adding" are used. There is no strict rule on the verb's form.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "yifan-c (via GitHub)" <gi...@apache.org>.
yifan-c commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1988085248
##########
client-common/src/main/java/org/apache/cassandra/sidecar/common/ApiEndpointsV1.java:
##########
@@ -121,13 +121,16 @@ public final class ApiEndpointsV1
public static final String LIST_CDC_SEGMENTS_ROUTE = API_V1 + CDC_PATH + "/segments";
public static final String STREAM_CDC_SEGMENTS_ROUTE = LIST_CDC_SEGMENTS_ROUTE + "/" + SEGMENT_PATH_PARAM;
+ // Schema Reporting
+ private static final String REPORT_SCHEMA = "/report-schema";
+ public static final String REPORT_SCHEMA_ROUTE = API_V1 + REPORT_SCHEMA;
Review Comment:
In a RESTful design, the resource in the plural form is a more popular convention. The API is to publishing/inserting a schema entity in the Datahub's `entities` resource. The intent of publishing is expressed by the HTTP verb (`PUT`. TBH, it should be `POST`, but I am fine with `PUT`), and the receiver is the resource collection, i.e. `schemas`. `datahub` in the API is to clarify the external service.
As mentioned in my previous comment, in DataHub's term, they are `entities`. But in the consideration of the sidecar clients, it is clearer with calling them `schemas`.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1988208028
##########
server/src/test/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandlerTest.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.io.IOException;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import com.datastax.driver.core.Metadata;
+import com.google.inject.AbstractModule;
+import com.google.inject.Guice;
+import com.google.inject.Injector;
+import com.google.inject.Provides;
+import com.google.inject.Singleton;
+import com.google.inject.util.Modules;
+import com.linkedin.mxe.MetadataChangeProposal;
+import datahub.client.Callback;
+import datahub.client.MetadataWriteResponse;
+import datahub.shaded.findbugs.annotations.SuppressFBWarnings;
+import io.netty.handler.codec.http.HttpResponseStatus;
+import io.vertx.core.Vertx;
+import io.vertx.ext.web.client.WebClient;
+import io.vertx.ext.web.client.predicate.ResponsePredicate;
+import io.vertx.junit5.VertxExtension;
+import io.vertx.junit5.VertxTestContext;
+import org.apache.cassandra.sidecar.TestModule;
+import org.apache.cassandra.sidecar.cluster.CassandraAdapterDelegate;
+import org.apache.cassandra.sidecar.cluster.InstancesMetadata;
+import org.apache.cassandra.sidecar.cluster.instance.InstanceMetadata;
+import org.apache.cassandra.sidecar.common.server.StorageOperations;
+import org.apache.cassandra.sidecar.common.server.utils.IOUtils;
+import org.apache.cassandra.sidecar.datahub.EmitterFactory;
+import org.apache.cassandra.sidecar.datahub.JsonEmitter;
+import org.apache.cassandra.sidecar.server.MainModule;
+import org.apache.cassandra.sidecar.server.Server;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link ReportSchemaHandler}
+ */
+@ExtendWith(VertxExtension.class)
+final class ReportSchemaHandlerTest
+{
+ private static final String CLUSTER = "cluster";
+ private static final String DIRECTORY = "/tmp";
+ private static final int IDENTIFIER = 42;
+ private static final String LOCALHOST = "127.0.0.1";
+ private static final int PORT = 9042;
+ private static final String ENDPOINT = "/api/v1/report-schema";
+ private static final Duration TIMEOUT = Duration.ofSeconds(5);
+
+ private static final class ThrowingEmitter extends JsonEmitter
+ {
+ @Override
+ @NotNull
+ public synchronized Future<MetadataWriteResponse> emit(@NotNull MetadataChangeProposal proposal,
+ @Nullable Callback callback) throws IOException
+ {
+ throw new IOException();
+ }
+ }
+
+ private final class ReportSchemaHandlerTestModule extends AbstractModule
+ {
+ @Provides
+ @Singleton
+ @NotNull
+ public InstancesMetadata instancesMetadata()
+ {
+ Metadata metadata = mock(Metadata.class);
+ when(metadata.getKeyspaces()).thenReturn(Collections.emptyList());
+
+ StorageOperations operations = mock(StorageOperations.class);
+ when(operations.clusterName()).thenReturn(CLUSTER);
+
+ CassandraAdapterDelegate delegate = mock(CassandraAdapterDelegate.class);
+ when(delegate.storageOperations()).thenReturn(operations);
+ when(delegate.metadata()).thenReturn(metadata);
+
+ InstanceMetadata instanceMetadata = mock(InstanceMetadata.class);
+ when(instanceMetadata.stagingDir()).thenReturn(DIRECTORY);
+ when(instanceMetadata.id()).thenReturn(IDENTIFIER);
+ when(instanceMetadata.host()).thenReturn(LOCALHOST);
+ when(instanceMetadata.port()).thenReturn(PORT);
+ when(instanceMetadata.delegate()).thenReturn(delegate);
+
+ InstancesMetadata instances = mock(InstancesMetadata.class);
+ when(instances.instances()).thenReturn(Collections.singletonList(instanceMetadata));
+ when(instances.instanceFromId(IDENTIFIER)).thenReturn(instanceMetadata);
+ when(instances.instanceFromHost(LOCALHOST)).thenReturn(instanceMetadata);
+ return instances;
+ }
+
+ @Provides
+ @Singleton
+ @NotNull
+ public EmitterFactory emitterFactory()
+ {
+ return () -> emitter;
+ }
+ }
+
+ private final Injector injector = Guice.createInjector(Modules.override(new MainModule()).with(
+ Modules.override(new TestModule()).with(
+ new ReportSchemaHandlerTestModule())));
+ private WebClient client;
+ private Server server;
+ private JsonEmitter emitter;
+
+ @BeforeEach
+ void before() throws InterruptedException
+ {
+ VertxTestContext context = new VertxTestContext();
+ server = injector.getInstance(Server.class);
+ server.start()
+ .onSuccess(result -> context.completeNow())
+ .onFailure(context::failNow);
+
+ client = WebClient.create(injector.getInstance(Vertx.class));
+
+ context.awaitCompletion(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @AfterEach
+ @SuppressFBWarnings("RV_RETURN_VALUE_IGNORED")
+ void after() throws InterruptedException
+ {
+ CountDownLatch latch = new CountDownLatch(1);
+ server.close()
+ .onSuccess(future -> latch.countDown());
+
+ client.close();
+
+ latch.await(TIMEOUT.toMillis(), TimeUnit.MILLISECONDS);
+ }
+
+ @Test
+ void testSuccess(@NotNull VertxTestContext context) throws IOException
+ {
+ String expected = IOUtils.readFully("/datahub/empty_cluster.json");
+ emitter = new JsonEmitter();
+ assertThat(emitter.content().length())
+ .isLessThanOrEqualTo(1);
+
+ client.put(server.actualPort(), LOCALHOST, ENDPOINT)
+ .expect(ResponsePredicate.SC_OK)
+ .send(context.succeeding(response ->
+ {
+ assertThat(response.statusCode())
+ .isEqualTo(HttpResponseStatus.OK.code());
+ assertThat(emitter.content())
+ .isEqualTo(expected);
+ context.completeNow();
+ }));
Review Comment:
Running `./gradlew idea` doesn't seem to update any files.
Just to confirm, you'd like me to apply automatic formatting to all files touched in this PR?
This will go against your desire to _keep the patch succinct, i.e. no unrelated changes_, as the formatting currently seems to be pretty inconsistent…
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1984015938
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandler.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.util.Collections;
+import java.util.Set;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.auth.authorization.Authorization;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.acl.authorization.BasicPermissions;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.datahub.SchemaReporter;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * An implementation of {@link AbstractHandler} used to trigger an immediate,
+ * synchronous conversion and report of the current schema
+ */
+@Singleton
+public class ReportSchemaHandler extends AbstractHandler<Void> implements AccessProtected
+{
+ @NotNull
+ private final SchemaReporter schemaReporter;
+
+ /**
+ * Constructs a new instance of {@link ReportSchemaHandler} using the provided instances
+ * of {@link InstanceMetadataFetcher}, {@link ExecutorPools}, and {@link SchemaReporter}
+ *
+ * @param metadata the metadata fetcher
+ * @param executor executor pools for blocking executions
+ * @param reporter executor pools for blocking executions
+ */
+ @Inject
+ public ReportSchemaHandler(@NotNull InstanceMetadataFetcher metadata,
+ @NotNull ExecutorPools executor,
+ @NotNull SchemaReporter reporter)
+ {
+ super(metadata, executor, null);
+
+ schemaReporter = reporter;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ @NotNull
+ public Set<Authorization> requiredAuthorizations()
+ {
+ return Collections.singleton(BasicPermissions.REPORT_SCHEMA.toAuthorization());
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ @Nullable
+ protected Void extractParamsOrThrow(@NotNull RoutingContext context)
+ {
+ return null;
+ }
+
+ /**
+ * {@inheritDoc}
+ */
+ @Override
+ protected void handleInternal(@NotNull RoutingContext context,
+ @NotNull HttpServerRequest http,
+ @NotNull String host,
+ @NotNull SocketAddress address,
+ @Nullable Void request)
+ {
+ executorPools.service()
+ .runBlocking(() -> metadataFetcher.runOnFirstAvailableInstance(instance ->
+ schemaReporter.process(instance.delegate().metadata())))
+ .onSuccess(context::json)
+ .onFailure(throwable -> processFailure(throwable, context, host, address, request));
+ }
Review Comment:
Very helpful idea fetching `Metadata` before scheduling on the pool, 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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1984012992
##########
server/src/main/java/org/apache/cassandra/sidecar/routes/ReportSchemaHandler.java:
##########
@@ -0,0 +1,101 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.sidecar.routes;
+
+import java.util.Collections;
+import java.util.Set;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import io.vertx.core.http.HttpServerRequest;
+import io.vertx.core.net.SocketAddress;
+import io.vertx.ext.auth.authorization.Authorization;
+import io.vertx.ext.web.RoutingContext;
+import org.apache.cassandra.sidecar.acl.authorization.BasicPermissions;
+import org.apache.cassandra.sidecar.concurrent.ExecutorPools;
+import org.apache.cassandra.sidecar.datahub.SchemaReporter;
+import org.apache.cassandra.sidecar.utils.InstanceMetadataFetcher;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * An implementation of {@link AbstractHandler} used to trigger an immediate,
+ * synchronous conversion and report of the current schema
+ */
+@Singleton
+public class ReportSchemaHandler extends AbstractHandler<Void> implements AccessProtected
+{
+ @NotNull
+ private final SchemaReporter schemaReporter;
+
+ /**
+ * Constructs a new instance of {@link ReportSchemaHandler} using the provided instances
+ * of {@link InstanceMetadataFetcher}, {@link ExecutorPools}, and {@link SchemaReporter}
+ *
+ * @param metadata the metadata fetcher
+ * @param executor executor pools for blocking executions
+ * @param reporter executor pools for blocking executions
Review Comment:
Fixed.
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "5 (via GitHub)" <gi...@apache.org>.
5 commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1984018801
##########
server/src/test/java/org/apache/cassandra/sidecar/utils/InstanceMetadataFetcherTest.java:
##########
@@ -21,7 +21,7 @@
import java.nio.file.Path;
import java.util.Arrays;
import java.util.List;
-
+import com.google.common.collect.ImmutableList;
Review Comment:
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "sarankk (via GitHub)" <gi...@apache.org>.
sarankk commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1964164599
##########
server/src/main/java/org/apache/cassandra/sidecar/acl/authorization/BasicPermissions.java:
##########
@@ -63,6 +63,9 @@ public class BasicPermissions
public static final Permission READ_OPERATIONAL_JOB = new DomainAwarePermission("OPERATIONAL_JOB:READ", OPERATION_SCOPE);
public static final Permission DECOMMISSION_NODE = new DomainAwarePermission("NODE:DECOMMISSION", OPERATION_SCOPE);
+ // Permissions related to Schema Reporting
+ public static final Permission REPORT_SCHEMA = new DomainAwarePermission("SCHEMA:REPORT", CLUSTER_SCOPE);
Review Comment:
I feel having 2 different permissions is better in general, with 1 granting read permission automatically grants them report permission. But in this particular case, since we already have a periodic task to report schema, irrespective of this endpoint, we can use the `enable` flag for schema reporting to control whether we want to allow reporting or not?
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org
Re: [PR] CASSSIDECAR-203: Created Endpoint that Triggers an Immediate Schema Report [cassandra-sidecar]
Posted by "frankgh (via GitHub)" <gi...@apache.org>.
frankgh commented on code in PR #198:
URL: https://github.com/apache/cassandra-sidecar/pull/198#discussion_r1964178304
##########
server/src/main/java/org/apache/cassandra/sidecar/acl/authorization/BasicPermissions.java:
##########
@@ -63,6 +63,9 @@ public class BasicPermissions
public static final Permission READ_OPERATIONAL_JOB = new DomainAwarePermission("OPERATIONAL_JOB:READ", OPERATION_SCOPE);
public static final Permission DECOMMISSION_NODE = new DomainAwarePermission("NODE:DECOMMISSION", OPERATION_SCOPE);
+ // Permissions related to Schema Reporting
+ public static final Permission REPORT_SCHEMA = new DomainAwarePermission("SCHEMA:REPORT", CLUSTER_SCOPE);
Review Comment:
I think we need a separate permission to allow a user trigger a schema reporting. We should not conflate the `SCHEMA:READ` permission with the ability to report schemas
--
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: pr-unsubscribe@cassandra.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: pr-unsubscribe@cassandra.apache.org
For additional commands, e-mail: pr-help@cassandra.apache.org