You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by GitBox <gi...@apache.org> on 2022/08/13 09:32:52 UTC
[GitHub] [inlong] healchow commented on a diff in pull request #5472: [INLONG-5047][Agent] Support collect data from SQLServer
healchow commented on code in PR #5472:
URL: https://github.com/apache/inlong/pull/5472#discussion_r945118352
##########
inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/sources/TestSQLServerSource.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.inlong.agent.plugin.sources;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.inlong.agent.conf.JobProfile;
+import org.apache.inlong.agent.constant.CommonConstants;
+import org.apache.inlong.agent.metrics.AgentMetricItem;
+import org.apache.inlong.agent.metrics.AgentMetricItemSet;
+import org.apache.inlong.common.metric.MetricItem;
+import org.apache.inlong.common.metric.MetricRegister;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.powermock.api.mockito.PowerMockito.when;
+import static org.powermock.api.mockito.PowerMockito.whenNew;
+import static org.powermock.api.support.membermodification.MemberMatcher.field;
+
+/**
+ * Test cases for {@link SQLServerSource}.
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({SQLServerSource.class, MetricRegister.class})
+@PowerMockIgnore({"javax.management.*"})
+public class TestSQLServerSource {
+
+ @Mock
+ JobProfile jobProfile;
+
+ @Mock
+ private AgentMetricItemSet agentMetricItemSet;
+
+ @Mock
+ private AgentMetricItem agentMetricItem;
+
+ private AtomicLong sourceSuccessCount;
+
+ private AtomicLong sourceFailCount;
+
+ @Before
+ public void setup() throws Exception {
+ sourceSuccessCount = new AtomicLong(0);
+ sourceFailCount = new AtomicLong(0);
+ //mock metrics
+ whenNew(AgentMetricItemSet.class).withArguments(anyString()).thenReturn(agentMetricItemSet);
+ when(agentMetricItemSet.findMetricItem(any())).thenReturn(agentMetricItem);
+ field(AgentMetricItem.class, "sourceSuccessCount").set(agentMetricItem, sourceSuccessCount);
+ field(AgentMetricItem.class, "sourceFailCount").set(agentMetricItem, sourceFailCount);
+ PowerMockito.mockStatic(MetricRegister.class);
+ PowerMockito.doNothing().when(
+ MetricRegister.class, "register", any(MetricItem.class));
+ }
+
+ /**
+ * Test cases for {@link SQLServerSource#split(JobProfile)}.
+ */
+ @Test
+ public void testSplit() {
+
+ final String sql1 = "select * from dbo.test01";
+ final String sql2 = "select * from dbo.test${01,99}";
+
+ // build mock
+ when(jobProfile.get(eq(CommonConstants.PROXY_INLONG_GROUP_ID), anyString())).thenReturn("test_group");
+ when(jobProfile.get(eq(CommonConstants.PROXY_INLONG_STREAM_ID), anyString())).thenReturn("test_stream");
+ when(jobProfile.get(eq(SQLServerSource.JOB_DATABASE_SQL), eq(StringUtils.EMPTY))).thenReturn(StringUtils.EMPTY,
+ sql1, sql2);
+
+ final SQLServerSource source = new SQLServerSource();
+
+ //assert
+ assertEquals(null, source.split(jobProfile));
Review Comment:
Add one blank after `//`.
##########
inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/sources/TestSQLServerReader.java:
##########
@@ -0,0 +1,257 @@
+/*
+ * 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.inlong.agent.plugin.sources;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.inlong.agent.conf.JobProfile;
+import org.apache.inlong.agent.constant.CommonConstants;
+import org.apache.inlong.agent.metrics.AgentMetricItem;
+import org.apache.inlong.agent.metrics.AgentMetricItemSet;
+import org.apache.inlong.agent.metrics.audit.AuditUtils;
+import org.apache.inlong.agent.plugin.Message;
+import org.apache.inlong.agent.plugin.Reader;
+import org.apache.inlong.agent.plugin.sources.reader.SQLServerReader;
+import org.apache.inlong.agent.utils.AgentDbUtils;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.Types;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.powermock.api.mockito.PowerMockito.field;
+import static org.powermock.api.mockito.PowerMockito.mockStatic;
+import static org.powermock.api.mockito.PowerMockito.when;
+import static org.powermock.api.mockito.PowerMockito.whenNew;
+
+/**
+ * Test cases for {@link SQLServerReader}.
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({AgentDbUtils.class, AuditUtils.class, SQLServerReader.class})
+@PowerMockIgnore({"javax.management.*"})
+public class TestSQLServerReader {
+
+ private SQLServerReader reader;
+
+ @Mock
+ private JobProfile jobProfile;
+
+ @Mock
+ private Connection conn;
+
+ @Mock
+ private PreparedStatement preparedStatement;
+
+ @Mock
+ private ResultSet resultSet;
+
+ @Mock
+ private ResultSetMetaData metaData;
+
+ @Mock
+ private AgentMetricItemSet agentMetricItemSet;
+
+ @Mock
+ private AgentMetricItem agentMetricItem;
+
+ private AtomicLong atomicLong;
+
+ private String tagName;
+
+ private String sql;
+
+ @Before
+ public void setUp() throws Exception {
+
+ final String username = "sa";
+ final String password = "123456";
+ final String hostname = "127.0.0.1";
+ final String port = "1434";
+ final String dbname = "inlong";
+ final String typeName1 = "int";
+ final String typeName2 = "varchar";
+ final String groupId = "group01";
+ final String streamId = "stream01";
+ atomicLong = new AtomicLong(0L);
+
+ sql = "select * from dbo.test01";
+ tagName = String.join("_", SQLServerReader.SQLSERVER_READER_TAG_NAME, groupId, streamId);
+
+ when(jobProfile.get(eq(CommonConstants.PROXY_INLONG_GROUP_ID), anyString())).thenReturn(groupId);
+ when(jobProfile.get(eq(CommonConstants.PROXY_INLONG_STREAM_ID), anyString())).thenReturn(streamId);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_USER))).thenReturn(username);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_PASSWORD))).thenReturn(password);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_HOSTNAME))).thenReturn(hostname);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_PORT))).thenReturn(port);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_DBNAME))).thenReturn(dbname);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_DRIVER_CLASS), anyString())).thenReturn(
+ SQLServerReader.DEFAULT_JOB_DATABASE_DRIVER_CLASS);
+ when(jobProfile.getInt(eq(SQLServerReader.JOB_DATABASE_BATCH_SIZE), anyInt())).thenReturn(
+ SQLServerReader.DEFAULT_JOB_DATABASE_BATCH_SIZE);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_TYPE), anyString())).thenReturn(
+ SQLServerReader.SQLSERVER);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_SEPARATOR), anyString())).thenReturn(
+ SQLServerReader.STD_FIELD_SEPARATOR_SHORT);
+ mockStatic(AgentDbUtils.class);
+ when(AgentDbUtils.getConnectionFailover(eq(SQLServerReader.DEFAULT_JOB_DATABASE_DRIVER_CLASS), anyString(),
+ eq(username), eq(password))).thenReturn(conn);
+ when(conn.prepareStatement(anyString())).thenReturn(preparedStatement);
+ when(preparedStatement.executeQuery()).thenReturn(resultSet);
+ when(resultSet.getMetaData()).thenReturn(metaData);
+ when(metaData.getColumnCount()).thenReturn(2);
+ when(metaData.getColumnName(1)).thenReturn("id");
+ when(metaData.getColumnName(2)).thenReturn("cell");
+ when(metaData.getColumnType(1)).thenReturn(Types.INTEGER);
+ when(metaData.getColumnType(2)).thenReturn(Types.VARCHAR);
+ when(metaData.getColumnTypeName(1)).thenReturn(typeName1);
+ when(metaData.getColumnTypeName(2)).thenReturn(typeName2);
+
+ //mock metrics
+ whenNew(AgentMetricItemSet.class).withArguments(anyString()).thenReturn(agentMetricItemSet);
+ when(agentMetricItemSet.findMetricItem(any())).thenReturn(agentMetricItem);
+ field(AgentMetricItem.class, "pluginReadCount").set(agentMetricItem, atomicLong);
+
+ //init method
+ reader = new SQLServerReader(sql);
+ reader.init(jobProfile);
+ }
+
+
+ /**
+ * Test cases for {@link SQLServerReader#read()}.
+ */
+ @Test
+ public void testRead() throws Exception {
+
Review Comment:
Unnecessary blank line.
##########
inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/sources/TestSQLServerSource.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.inlong.agent.plugin.sources;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.inlong.agent.conf.JobProfile;
+import org.apache.inlong.agent.constant.CommonConstants;
+import org.apache.inlong.agent.metrics.AgentMetricItem;
+import org.apache.inlong.agent.metrics.AgentMetricItemSet;
+import org.apache.inlong.common.metric.MetricItem;
+import org.apache.inlong.common.metric.MetricRegister;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.powermock.api.mockito.PowerMockito;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.powermock.api.mockito.PowerMockito.when;
+import static org.powermock.api.mockito.PowerMockito.whenNew;
+import static org.powermock.api.support.membermodification.MemberMatcher.field;
+
+/**
+ * Test cases for {@link SQLServerSource}.
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({SQLServerSource.class, MetricRegister.class})
+@PowerMockIgnore({"javax.management.*"})
+public class TestSQLServerSource {
+
+ @Mock
+ JobProfile jobProfile;
+
+ @Mock
+ private AgentMetricItemSet agentMetricItemSet;
+
+ @Mock
+ private AgentMetricItem agentMetricItem;
+
+ private AtomicLong sourceSuccessCount;
+
+ private AtomicLong sourceFailCount;
+
+ @Before
+ public void setup() throws Exception {
+ sourceSuccessCount = new AtomicLong(0);
+ sourceFailCount = new AtomicLong(0);
+ //mock metrics
+ whenNew(AgentMetricItemSet.class).withArguments(anyString()).thenReturn(agentMetricItemSet);
+ when(agentMetricItemSet.findMetricItem(any())).thenReturn(agentMetricItem);
+ field(AgentMetricItem.class, "sourceSuccessCount").set(agentMetricItem, sourceSuccessCount);
+ field(AgentMetricItem.class, "sourceFailCount").set(agentMetricItem, sourceFailCount);
+ PowerMockito.mockStatic(MetricRegister.class);
+ PowerMockito.doNothing().when(
+ MetricRegister.class, "register", any(MetricItem.class));
+ }
+
+ /**
+ * Test cases for {@link SQLServerSource#split(JobProfile)}.
+ */
+ @Test
+ public void testSplit() {
+
Review Comment:
Unnecessary blank line.
##########
inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/sources/TestSQLServerReader.java:
##########
@@ -0,0 +1,257 @@
+/*
+ * 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.inlong.agent.plugin.sources;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.inlong.agent.conf.JobProfile;
+import org.apache.inlong.agent.constant.CommonConstants;
+import org.apache.inlong.agent.metrics.AgentMetricItem;
+import org.apache.inlong.agent.metrics.AgentMetricItemSet;
+import org.apache.inlong.agent.metrics.audit.AuditUtils;
+import org.apache.inlong.agent.plugin.Message;
+import org.apache.inlong.agent.plugin.Reader;
+import org.apache.inlong.agent.plugin.sources.reader.SQLServerReader;
+import org.apache.inlong.agent.utils.AgentDbUtils;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.Types;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.powermock.api.mockito.PowerMockito.field;
+import static org.powermock.api.mockito.PowerMockito.mockStatic;
+import static org.powermock.api.mockito.PowerMockito.when;
+import static org.powermock.api.mockito.PowerMockito.whenNew;
+
+/**
+ * Test cases for {@link SQLServerReader}.
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({AgentDbUtils.class, AuditUtils.class, SQLServerReader.class})
+@PowerMockIgnore({"javax.management.*"})
+public class TestSQLServerReader {
+
+ private SQLServerReader reader;
+
+ @Mock
+ private JobProfile jobProfile;
+
+ @Mock
+ private Connection conn;
+
+ @Mock
+ private PreparedStatement preparedStatement;
+
+ @Mock
+ private ResultSet resultSet;
+
+ @Mock
+ private ResultSetMetaData metaData;
+
+ @Mock
+ private AgentMetricItemSet agentMetricItemSet;
+
+ @Mock
+ private AgentMetricItem agentMetricItem;
+
+ private AtomicLong atomicLong;
+
+ private String tagName;
+
+ private String sql;
+
+ @Before
+ public void setUp() throws Exception {
+
+ final String username = "sa";
+ final String password = "123456";
+ final String hostname = "127.0.0.1";
+ final String port = "1434";
+ final String dbname = "inlong";
+ final String typeName1 = "int";
+ final String typeName2 = "varchar";
+ final String groupId = "group01";
+ final String streamId = "stream01";
+ atomicLong = new AtomicLong(0L);
+
+ sql = "select * from dbo.test01";
+ tagName = String.join("_", SQLServerReader.SQLSERVER_READER_TAG_NAME, groupId, streamId);
+
+ when(jobProfile.get(eq(CommonConstants.PROXY_INLONG_GROUP_ID), anyString())).thenReturn(groupId);
+ when(jobProfile.get(eq(CommonConstants.PROXY_INLONG_STREAM_ID), anyString())).thenReturn(streamId);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_USER))).thenReturn(username);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_PASSWORD))).thenReturn(password);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_HOSTNAME))).thenReturn(hostname);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_PORT))).thenReturn(port);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_DBNAME))).thenReturn(dbname);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_DRIVER_CLASS), anyString())).thenReturn(
+ SQLServerReader.DEFAULT_JOB_DATABASE_DRIVER_CLASS);
+ when(jobProfile.getInt(eq(SQLServerReader.JOB_DATABASE_BATCH_SIZE), anyInt())).thenReturn(
+ SQLServerReader.DEFAULT_JOB_DATABASE_BATCH_SIZE);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_TYPE), anyString())).thenReturn(
+ SQLServerReader.SQLSERVER);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_SEPARATOR), anyString())).thenReturn(
+ SQLServerReader.STD_FIELD_SEPARATOR_SHORT);
+ mockStatic(AgentDbUtils.class);
+ when(AgentDbUtils.getConnectionFailover(eq(SQLServerReader.DEFAULT_JOB_DATABASE_DRIVER_CLASS), anyString(),
+ eq(username), eq(password))).thenReturn(conn);
+ when(conn.prepareStatement(anyString())).thenReturn(preparedStatement);
+ when(preparedStatement.executeQuery()).thenReturn(resultSet);
+ when(resultSet.getMetaData()).thenReturn(metaData);
+ when(metaData.getColumnCount()).thenReturn(2);
+ when(metaData.getColumnName(1)).thenReturn("id");
+ when(metaData.getColumnName(2)).thenReturn("cell");
+ when(metaData.getColumnType(1)).thenReturn(Types.INTEGER);
+ when(metaData.getColumnType(2)).thenReturn(Types.VARCHAR);
+ when(metaData.getColumnTypeName(1)).thenReturn(typeName1);
+ when(metaData.getColumnTypeName(2)).thenReturn(typeName2);
+
+ //mock metrics
Review Comment:
Add one blank after `//`.
##########
inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/sources/TestSQLServerReader.java:
##########
@@ -0,0 +1,257 @@
+/*
+ * 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.inlong.agent.plugin.sources;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.inlong.agent.conf.JobProfile;
+import org.apache.inlong.agent.constant.CommonConstants;
+import org.apache.inlong.agent.metrics.AgentMetricItem;
+import org.apache.inlong.agent.metrics.AgentMetricItemSet;
+import org.apache.inlong.agent.metrics.audit.AuditUtils;
+import org.apache.inlong.agent.plugin.Message;
+import org.apache.inlong.agent.plugin.Reader;
+import org.apache.inlong.agent.plugin.sources.reader.SQLServerReader;
+import org.apache.inlong.agent.utils.AgentDbUtils;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.Types;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.powermock.api.mockito.PowerMockito.field;
+import static org.powermock.api.mockito.PowerMockito.mockStatic;
+import static org.powermock.api.mockito.PowerMockito.when;
+import static org.powermock.api.mockito.PowerMockito.whenNew;
+
+/**
+ * Test cases for {@link SQLServerReader}.
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({AgentDbUtils.class, AuditUtils.class, SQLServerReader.class})
+@PowerMockIgnore({"javax.management.*"})
+public class TestSQLServerReader {
+
+ private SQLServerReader reader;
+
+ @Mock
+ private JobProfile jobProfile;
+
+ @Mock
+ private Connection conn;
+
+ @Mock
+ private PreparedStatement preparedStatement;
+
+ @Mock
+ private ResultSet resultSet;
+
+ @Mock
+ private ResultSetMetaData metaData;
+
+ @Mock
+ private AgentMetricItemSet agentMetricItemSet;
+
+ @Mock
+ private AgentMetricItem agentMetricItem;
+
+ private AtomicLong atomicLong;
+
+ private String tagName;
+
+ private String sql;
+
+ @Before
+ public void setUp() throws Exception {
+
+ final String username = "sa";
+ final String password = "123456";
+ final String hostname = "127.0.0.1";
+ final String port = "1434";
+ final String dbname = "inlong";
+ final String typeName1 = "int";
+ final String typeName2 = "varchar";
+ final String groupId = "group01";
+ final String streamId = "stream01";
+ atomicLong = new AtomicLong(0L);
+
+ sql = "select * from dbo.test01";
+ tagName = String.join("_", SQLServerReader.SQLSERVER_READER_TAG_NAME, groupId, streamId);
+
+ when(jobProfile.get(eq(CommonConstants.PROXY_INLONG_GROUP_ID), anyString())).thenReturn(groupId);
+ when(jobProfile.get(eq(CommonConstants.PROXY_INLONG_STREAM_ID), anyString())).thenReturn(streamId);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_USER))).thenReturn(username);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_PASSWORD))).thenReturn(password);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_HOSTNAME))).thenReturn(hostname);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_PORT))).thenReturn(port);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_DBNAME))).thenReturn(dbname);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_DRIVER_CLASS), anyString())).thenReturn(
+ SQLServerReader.DEFAULT_JOB_DATABASE_DRIVER_CLASS);
+ when(jobProfile.getInt(eq(SQLServerReader.JOB_DATABASE_BATCH_SIZE), anyInt())).thenReturn(
+ SQLServerReader.DEFAULT_JOB_DATABASE_BATCH_SIZE);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_TYPE), anyString())).thenReturn(
+ SQLServerReader.SQLSERVER);
+ when(jobProfile.get(eq(SQLServerReader.JOB_DATABASE_SEPARATOR), anyString())).thenReturn(
+ SQLServerReader.STD_FIELD_SEPARATOR_SHORT);
+ mockStatic(AgentDbUtils.class);
+ when(AgentDbUtils.getConnectionFailover(eq(SQLServerReader.DEFAULT_JOB_DATABASE_DRIVER_CLASS), anyString(),
+ eq(username), eq(password))).thenReturn(conn);
+ when(conn.prepareStatement(anyString())).thenReturn(preparedStatement);
+ when(preparedStatement.executeQuery()).thenReturn(resultSet);
+ when(resultSet.getMetaData()).thenReturn(metaData);
+ when(metaData.getColumnCount()).thenReturn(2);
+ when(metaData.getColumnName(1)).thenReturn("id");
+ when(metaData.getColumnName(2)).thenReturn("cell");
+ when(metaData.getColumnType(1)).thenReturn(Types.INTEGER);
+ when(metaData.getColumnType(2)).thenReturn(Types.VARCHAR);
+ when(metaData.getColumnTypeName(1)).thenReturn(typeName1);
+ when(metaData.getColumnTypeName(2)).thenReturn(typeName2);
+
+ //mock metrics
+ whenNew(AgentMetricItemSet.class).withArguments(anyString()).thenReturn(agentMetricItemSet);
+ when(agentMetricItemSet.findMetricItem(any())).thenReturn(agentMetricItem);
+ field(AgentMetricItem.class, "pluginReadCount").set(agentMetricItem, atomicLong);
+
+ //init method
+ reader = new SQLServerReader(sql);
+ reader.init(jobProfile);
+ }
+
Review Comment:
Unnecessary more than one line.
##########
inlong-agent/agent-plugins/src/test/java/org/apache/inlong/agent/plugin/sources/TestSQLServerReader.java:
##########
@@ -0,0 +1,257 @@
+/*
+ * 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.inlong.agent.plugin.sources;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.inlong.agent.conf.JobProfile;
+import org.apache.inlong.agent.constant.CommonConstants;
+import org.apache.inlong.agent.metrics.AgentMetricItem;
+import org.apache.inlong.agent.metrics.AgentMetricItemSet;
+import org.apache.inlong.agent.metrics.audit.AuditUtils;
+import org.apache.inlong.agent.plugin.Message;
+import org.apache.inlong.agent.plugin.Reader;
+import org.apache.inlong.agent.plugin.sources.reader.SQLServerReader;
+import org.apache.inlong.agent.utils.AgentDbUtils;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.powermock.core.classloader.annotations.PowerMockIgnore;
+import org.powermock.core.classloader.annotations.PrepareForTest;
+import org.powermock.modules.junit4.PowerMockRunner;
+
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.Types;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicLong;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.powermock.api.mockito.PowerMockito.field;
+import static org.powermock.api.mockito.PowerMockito.mockStatic;
+import static org.powermock.api.mockito.PowerMockito.when;
+import static org.powermock.api.mockito.PowerMockito.whenNew;
+
+/**
+ * Test cases for {@link SQLServerReader}.
+ */
+@RunWith(PowerMockRunner.class)
+@PrepareForTest({AgentDbUtils.class, AuditUtils.class, SQLServerReader.class})
+@PowerMockIgnore({"javax.management.*"})
+public class TestSQLServerReader {
+
+ private SQLServerReader reader;
+
+ @Mock
+ private JobProfile jobProfile;
+
+ @Mock
+ private Connection conn;
+
+ @Mock
+ private PreparedStatement preparedStatement;
+
+ @Mock
+ private ResultSet resultSet;
+
+ @Mock
+ private ResultSetMetaData metaData;
+
+ @Mock
+ private AgentMetricItemSet agentMetricItemSet;
+
+ @Mock
+ private AgentMetricItem agentMetricItem;
+
+ private AtomicLong atomicLong;
+
+ private String tagName;
+
+ private String sql;
+
+ @Before
+ public void setUp() throws Exception {
+
Review Comment:
Unnecessary blank line.
##########
inlong-agent/agent-plugins/src/main/java/org/apache/inlong/agent/plugin/sources/reader/SQLServerReader.java:
##########
@@ -0,0 +1,222 @@
+/*
+ * 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.inlong.agent.plugin.sources.reader;
+
+import org.apache.commons.codec.binary.Base64;
+import org.apache.commons.lang3.CharUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.inlong.agent.conf.JobProfile;
+import org.apache.inlong.agent.message.DefaultMessage;
+import org.apache.inlong.agent.metrics.audit.AuditUtils;
+import org.apache.inlong.agent.plugin.Message;
+import org.apache.inlong.agent.utils.AgentDbUtils;
+import org.apache.inlong.agent.utils.AgentUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.nio.charset.StandardCharsets;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.util.ArrayList;
+import java.util.List;
+
+import static java.sql.Types.BINARY;
+import static java.sql.Types.BLOB;
+import static java.sql.Types.LONGVARBINARY;
+import static java.sql.Types.VARBINARY;
+
+public class SQLServerReader extends AbstractReader {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SqlReader.class);
+
+ public static final String SQLSERVER_READER_TAG_NAME = "AgentSQLServerMetric";
+
+ public static final String JOB_DATABASE_USER = "job.sqlserverJob.user";
+ public static final String JOB_DATABASE_PASSWORD = "job.sqlserverJob.password";
+ public static final String JOB_DATABASE_HOSTNAME = "job.sqlserverJob.hostname";
+ public static final String JOB_DATABASE_PORT = "job.sqlserverJob.port";
+ public static final String JOB_DATABASE_DBNAME = "job.sqlserverJob.dbname";
+
+ public static final String JOB_DATABASE_BATCH_SIZE = "job.sqlserverJob.batchSize";
+ public static final int DEFAULT_JOB_DATABASE_BATCH_SIZE = 1000;
+
+ public static final String JOB_DATABASE_DRIVER_CLASS = "job.database.driverClass";
+ public static final String DEFAULT_JOB_DATABASE_DRIVER_CLASS = "com.microsoft.sqlserver.jdbc.SQLServerDriver";
+
+ public static final String STD_FIELD_SEPARATOR_SHORT = "\001";
+ public static final String JOB_DATABASE_SEPARATOR = "job.sql.separator";
+
+ // pre-set sql lines, commands like "set xxx=xx;"
+ public static final String JOB_DATABASE_TYPE = "job.database.type";
+ public static final String SQLSERVER = "sqlserver";
+
+ private static final String[] NEW_LINE_CHARS = new String[]{String.valueOf(CharUtils.CR),
+ String.valueOf(CharUtils.LF)};
+ private static final String[] EMPTY_CHARS = new String[]{StringUtils.EMPTY, StringUtils.EMPTY};
+
+ private final String sql;
+
+ private PreparedStatement preparedStatement;
+ private Connection conn;
+ private ResultSet resultSet;
+ private int columnCount;
+
+ // column types
+ private String[] columnTypeNames;
+ private int[] columnTypeCodes;
+ private boolean finished = false;
+ private String separator;
+
+ public SQLServerReader(String sql) {
+ this.sql = sql;
+ }
+
+ @Override
+ public Message read() {
+ try {
+ if (resultSet.next()) {
Review Comment:
Is it possible to use guard statements to reduce the nesting of code here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@inlong.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org